@@ -10,6 +10,7 @@ use std::{
1010 collections:: { hash_map:: DefaultHasher , HashMap } ,
1111 hash:: { Hash , Hasher } ,
1212 marker:: PhantomData ,
13+ time:: { Duration , Instant } ,
1314} ;
1415
1516fn hash < T : Hash + ?Sized > ( t : & T ) -> u64 {
@@ -114,6 +115,44 @@ where
114115 }
115116}
116117
118+ /// Configuration for predicate filtering with cache TTL
119+ #[ derive( Debug , Clone ) ]
120+ pub struct Config {
121+ /// Time-to-live for cache entries
122+ ///
123+ /// Entries not seen for at least this long will be evicted from the cache.
124+ /// Default is 1 hour.
125+ ttl : Duration ,
126+ }
127+
128+ impl Config {
129+ /// Set the time-to-live for cache entries
130+ ///
131+ /// Entries not seen for at least this long will be evicted from the cache.
132+ #[ must_use]
133+ pub fn ttl ( mut self , ttl : Duration ) -> Self {
134+ self . ttl = ttl;
135+ self
136+ }
137+ }
138+
139+ impl Default for Config {
140+ fn default ( ) -> Self {
141+ Self {
142+ // Default to 1 hour TTL - long enough to avoid unnecessary reconciles
143+ // but short enough to prevent unbounded memory growth
144+ ttl : Duration :: from_secs ( 3600 ) ,
145+ }
146+ }
147+ }
148+
149+ /// Cache entry storing predicate hash and last access time
150+ #[ derive( Debug , Clone ) ]
151+ struct CacheEntry {
152+ hash : u64 ,
153+ last_seen : Instant ,
154+ }
155+
117156#[ allow( clippy:: pedantic) ]
118157#[ pin_project]
119158/// Stream returned by the [`predicate_filter`](super::WatchStreamExt::predicate_filter) method.
@@ -122,7 +161,8 @@ pub struct PredicateFilter<St, K: Resource, P: Predicate<K>> {
122161 #[ pin]
123162 stream : St ,
124163 predicate : P ,
125- cache : HashMap < PredicateCacheKey , u64 > ,
164+ cache : HashMap < PredicateCacheKey , CacheEntry > ,
165+ config : Config ,
126166 // K: Resource necessary to get .meta() of an object during polling
127167 _phantom : PhantomData < K > ,
128168}
@@ -132,11 +172,12 @@ where
132172 K : Resource ,
133173 P : Predicate < K > ,
134174{
135- pub ( super ) fn new ( stream : St , predicate : P ) -> Self {
175+ pub ( super ) fn new ( stream : St , predicate : P , config : Config ) -> Self {
136176 Self {
137177 stream,
138178 predicate,
139179 cache : HashMap :: new ( ) ,
180+ config,
140181 _phantom : PhantomData ,
141182 }
142183 }
@@ -152,13 +193,29 @@ where
152193
153194 fn poll_next ( self : Pin < & mut Self > , cx : & mut Context < ' _ > ) -> Poll < Option < Self :: Item > > {
154195 let mut me = self . project ( ) ;
196+
197+ // Evict expired entries before processing new events
198+ let now = Instant :: now ( ) ;
199+ let ttl = me. config . ttl ;
200+ me. cache
201+ . retain ( |_, entry| now. duration_since ( entry. last_seen ) < ttl) ;
202+
155203 Poll :: Ready ( loop {
156204 break match ready ! ( me. stream. as_mut( ) . poll_next( cx) ) {
157205 Some ( Ok ( obj) ) => {
158206 if let Some ( val) = me. predicate . hash_property ( & obj) {
159207 let key = PredicateCacheKey :: from ( obj. meta ( ) ) ;
160- let changed = me. cache . get ( & key) != Some ( & val) ;
161- me. cache . insert ( key, val) ;
208+ let now = Instant :: now ( ) ;
209+
210+ // Check if the predicate value changed or entry doesn't exist
211+ let changed = me. cache . get ( & key) . map ( |entry| entry. hash ) != Some ( val) ;
212+
213+ // Upsert the cache entry with new hash and timestamp
214+ me. cache . insert ( key, CacheEntry {
215+ hash : val,
216+ last_seen : now,
217+ } ) ;
218+
162219 if changed {
163220 Some ( Ok ( obj) )
164221 } else {
@@ -216,7 +273,7 @@ pub mod predicates {
216273pub ( crate ) mod tests {
217274 use std:: { pin:: pin, task:: Poll } ;
218275
219- use super :: { predicates, Error , PredicateFilter } ;
276+ use super :: { predicates, Config , Error , PredicateFilter } ;
220277 use futures:: { poll, stream, FutureExt , StreamExt } ;
221278 use kube_client:: Resource ;
222279 use serde_json:: json;
@@ -248,7 +305,11 @@ pub(crate) mod tests {
248305 Ok ( mkobj ( 1 ) ) ,
249306 Ok ( mkobj ( 2 ) ) ,
250307 ] ) ;
251- let mut rx = pin ! ( PredicateFilter :: new( data, predicates:: generation) ) ;
308+ let mut rx = pin ! ( PredicateFilter :: new(
309+ data,
310+ predicates:: generation,
311+ Config :: default ( )
312+ ) ) ;
252313
253314 // mkobj(1) passed through
254315 let first = rx. next ( ) . now_or_never ( ) . unwrap ( ) . unwrap ( ) . unwrap ( ) ;
@@ -299,7 +360,11 @@ pub(crate) mod tests {
299360 Ok ( mkobj ( 1 , "uid-2" ) ) ,
300361 Ok ( mkobj ( 2 , "uid-3" ) ) ,
301362 ] ) ;
302- let mut rx = pin ! ( PredicateFilter :: new( data, predicates:: generation) ) ;
363+ let mut rx = pin ! ( PredicateFilter :: new(
364+ data,
365+ predicates:: generation,
366+ Config :: default ( )
367+ ) ) ;
303368
304369 // mkobj(1, uid-1) passed through
305370 let first = rx. next ( ) . now_or_never ( ) . unwrap ( ) . unwrap ( ) . unwrap ( ) ;
@@ -319,4 +384,60 @@ pub(crate) mod tests {
319384
320385 assert ! ( matches!( poll!( rx. next( ) ) , Poll :: Ready ( None ) ) ) ;
321386 }
387+
388+ #[ tokio:: test]
389+ async fn predicate_cache_ttl_evicts_expired_entries ( ) {
390+ use futures:: { channel:: mpsc, SinkExt } ;
391+ use k8s_openapi:: api:: core:: v1:: Pod ;
392+ use std:: time:: Duration ;
393+
394+ let mkobj = |g : i32 , uid : & str | {
395+ let p: Pod = serde_json:: from_value ( json ! ( {
396+ "apiVersion" : "v1" ,
397+ "kind" : "Pod" ,
398+ "metadata" : {
399+ "name" : "blog" ,
400+ "namespace" : "default" ,
401+ "generation" : Some ( g) ,
402+ "uid" : uid,
403+ } ,
404+ "spec" : {
405+ "containers" : [ {
406+ "name" : "blog" ,
407+ "image" : "clux/blog:0.1.0"
408+ } ] ,
409+ }
410+ } ) )
411+ . unwrap ( ) ;
412+ p
413+ } ;
414+
415+ // Use a very short TTL for testing
416+ let config = Config :: default ( ) . ttl ( Duration :: from_millis ( 50 ) ) ;
417+
418+ // Use a channel so we can send items with delays
419+ let ( mut tx, rx) = mpsc:: unbounded ( ) ;
420+ let mut filtered = pin ! ( PredicateFilter :: new(
421+ rx. map( Ok :: <_, Error >) ,
422+ predicates:: generation,
423+ config
424+ ) ) ;
425+
426+ // Send first object
427+ tx. send ( mkobj ( 1 , "uid-1" ) ) . await . unwrap ( ) ;
428+ let first = filtered. next ( ) . now_or_never ( ) . unwrap ( ) . unwrap ( ) . unwrap ( ) ;
429+ assert_eq ! ( first. meta( ) . generation, Some ( 1 ) ) ;
430+
431+ // Send same object immediately - should be filtered
432+ tx. send ( mkobj ( 1 , "uid-1" ) ) . await . unwrap ( ) ;
433+ assert ! ( matches!( poll!( filtered. next( ) ) , Poll :: Pending ) ) ;
434+
435+ // Wait for TTL to expire
436+ tokio:: time:: sleep ( Duration :: from_millis ( 100 ) ) . await ;
437+
438+ // Send same object after TTL - should pass through due to eviction
439+ tx. send ( mkobj ( 1 , "uid-1" ) ) . await . unwrap ( ) ;
440+ let second = filtered. next ( ) . now_or_never ( ) . unwrap ( ) . unwrap ( ) . unwrap ( ) ;
441+ assert_eq ! ( second. meta( ) . generation, Some ( 1 ) ) ;
442+ }
322443}
0 commit comments