@@ -4,12 +4,13 @@ use anyhow::Result;
44use std:: collections:: { HashMap , HashSet } ;
55use std:: sync:: Arc ;
66use std:: time:: Instant ;
7- use tracing:: { debug, warn} ;
7+ use tokio:: sync:: { broadcast, RwLock } ;
8+ use tracing:: { debug, info, warn} ;
89
910use crate :: services:: PrefixCacheService ;
1011use mcpmux_core:: {
11- FeatureSet , FeatureSetRepository , FeatureType , MemberMode , MemberType , ServerFeature ,
12- ServerFeatureRepository ,
12+ DomainEvent , FeatureSet , FeatureSetRepository , FeatureType , MemberMode , MemberType ,
13+ ServerFeature , ServerFeatureRepository ,
1314} ;
1415
1516/// A catalog tool visible in discovery but not invokable until its FeatureSet is bound.
@@ -32,11 +33,18 @@ fn apply_mode_to_set(
3233 }
3334}
3435
36+ /// Cache key: space + sorted FeatureSet ids. Type filter is applied after
37+ /// the hit so tools/prompts/resources share one entry.
38+ type ResolutionCacheKey = ( String , Vec < String > ) ;
39+
3540/// Handles feature set resolution and permission evaluation
3641pub struct FeatureResolutionService {
3742 feature_repo : Arc < dyn ServerFeatureRepository > ,
3843 feature_set_repo : Arc < dyn FeatureSetRepository > ,
3944 prefix_cache : Arc < PrefixCacheService > ,
45+ /// Resolved (allow/exclude + prefix) features, invalidated when
46+ /// [`DomainEvent::affects_mcp_capabilities`] is true.
47+ cache : Arc < RwLock < HashMap < ResolutionCacheKey , Vec < ServerFeature > > > > ,
4048}
4149
4250impl FeatureResolutionService {
@@ -49,9 +57,61 @@ impl FeatureResolutionService {
4957 feature_repo,
5058 feature_set_repo,
5159 prefix_cache,
60+ cache : Arc :: new ( RwLock :: new ( HashMap :: new ( ) ) ) ,
5261 }
5362 }
5463
64+ /// Drop cached resolutions when a capability-changing domain event fires.
65+ ///
66+ /// Uses the same [`DomainEvent::affects_mcp_capabilities`] predicate as
67+ /// [`crate::consumers::MCPNotifier`] so invalidation stays in lockstep
68+ /// with `list_changed` fanout. Space-scoped when the event carries a
69+ /// `space_id`; whole-cache drop on lag (missed events).
70+ pub fn start_cache_invalidation (
71+ self : Arc < Self > ,
72+ mut event_rx : broadcast:: Receiver < DomainEvent > ,
73+ ) {
74+ tokio:: spawn ( async move {
75+ info ! ( "[FeatureResolution] cache invalidation listener started" ) ;
76+ loop {
77+ match event_rx. recv ( ) . await {
78+ Ok ( event) => {
79+ if !event. affects_mcp_capabilities ( ) {
80+ continue ;
81+ }
82+ if let Some ( space_id) = event. space_id ( ) {
83+ self . invalidate_space ( & space_id. to_string ( ) ) . await ;
84+ } else {
85+ self . invalidate_all ( ) . await ;
86+ }
87+ }
88+ Err ( broadcast:: error:: RecvError :: Lagged ( skipped) ) => {
89+ warn ! (
90+ skipped,
91+ "[FeatureResolution] lagged — dropping resolution cache"
92+ ) ;
93+ self . invalidate_all ( ) . await ;
94+ }
95+ Err ( broadcast:: error:: RecvError :: Closed ) => {
96+ warn ! ( "[FeatureResolution] event channel closed, stopping cache listener" ) ;
97+ break ;
98+ }
99+ }
100+ }
101+ } ) ;
102+ }
103+
104+ async fn invalidate_space ( & self , space_id : & str ) {
105+ self . cache
106+ . write ( )
107+ . await
108+ . retain ( |( cached_space, _) , _| cached_space != space_id) ;
109+ }
110+
111+ async fn invalidate_all ( & self ) {
112+ self . cache . write ( ) . await . clear ( ) ;
113+ }
114+
55115 /// Get all available features for a space (optionally filtered by type)
56116pub async fn get_all_features_for_space (
57117 & self ,
@@ -87,6 +147,38 @@ impl FeatureResolutionService {
87147 space_id : & str ,
88148 feature_set_ids : & [ String ] ,
89149 filter_type : Option < FeatureType > ,
150+ ) -> Result < Vec < ServerFeature > > {
151+ let mut sorted_ids = feature_set_ids. to_vec ( ) ;
152+ sorted_ids. sort ( ) ;
153+ let key = ( space_id. to_string ( ) , sorted_ids) ;
154+
155+ if let Some ( cached) = self . cache . read ( ) . await . get ( & key) . cloned ( ) {
156+ return Ok ( Self :: apply_type_filter ( cached, filter_type) ) ;
157+ }
158+
159+ // ponytail: concurrent misses recompute; single-flight if cold-start
160+ // stampede shows up.
161+ let resolved = self
162+ . resolve_feature_sets_uncached ( space_id, feature_set_ids)
163+ . await ?;
164+ self . cache . write ( ) . await . insert ( key, resolved. clone ( ) ) ;
165+ Ok ( Self :: apply_type_filter ( resolved, filter_type) )
166+ }
167+
168+ fn apply_type_filter (
169+ mut features : Vec < ServerFeature > ,
170+ filter_type : Option < FeatureType > ,
171+ ) -> Vec < ServerFeature > {
172+ if let Some ( feature_type) = filter_type {
173+ features. retain ( |f| f. feature_type == feature_type) ;
174+ }
175+ features
176+ }
177+
178+ async fn resolve_feature_sets_uncached (
179+ & self ,
180+ space_id : & str ,
181+ feature_set_ids : & [ String ] ,
90182 ) -> Result < Vec < ServerFeature > > {
91183 let mut allowed_feature_ids: HashSet < String > = HashSet :: new ( ) ;
92184 let mut excluded_feature_ids: HashSet < String > = HashSet :: new ( ) ;
@@ -144,33 +236,26 @@ impl FeatureResolutionService {
144236 excluded_feature_ids. len( )
145237 ) ;
146238
239+ let mut filtered_out = 0usize ;
147240 let mut result: Vec < ServerFeature > = all_features
148241 . into_iter ( )
149242 . filter ( |f| {
150243 let in_allowed = allowed_feature_ids. contains ( & f. id . to_string ( ) ) ;
151244 let in_excluded = excluded_feature_ids. contains ( & f. id . to_string ( ) ) ;
152245 let passes = f. is_available && in_allowed && !in_excluded;
153246 if !passes && in_allowed {
154- debug ! (
155- "[FeatureResolution] Feature {} (server={}) filtered out: is_available={}, in_allowed={}, in_excluded={}" ,
156- f. feature_name, f. server_id, f. is_available, in_allowed, in_excluded
157- ) ;
247+ filtered_out += 1 ;
158248 }
159249 passes
160250 } )
161251 . collect ( ) ;
162252
163253 debug ! (
164- "[FeatureResolution] After filter: {} features" ,
165- result. len( )
254+ "[FeatureResolution] After filter: {} features, filtered_out={}" ,
255+ result. len( ) ,
256+ filtered_out
166257 ) ;
167258
168- // Apply type filter if specified (OCP)
169- if let Some ( feature_type) = filter_type {
170- result. retain ( |f| f. feature_type == feature_type) ;
171- }
172-
173- // Enrich with prefixes
174259 for feature in & mut result {
175260 let prefix = self
176261 . prefix_cache
0 commit comments