multi_tier_cache/
invalidation.rs

1//! Cache invalidation and synchronization module
2//!
3//! This module provides cross-instance cache invalidation using Redis Pub/Sub.
4//! It supports both cache removal (invalidation) and cache updates (refresh).
5
6use crate::error::CacheResult;
7use crate::traits::StreamingBackend;
8use bytes::Bytes;
9use futures_util::StreamExt;
10use redis::AsyncCommands;
11use serde::{Deserialize, Serialize};
12use std::time::{Duration, SystemTime, UNIX_EPOCH};
13use tokio::sync::broadcast;
14use tracing::{debug, error, info, warn};
15use uuid::Uuid;
16
17/// Invalidation message types sent across cache instances via Redis Pub/Sub
18#[derive(Debug, Clone, Serialize, Deserialize)]
19#[serde(tag = "type")]
20pub enum InvalidationMessage {
21    /// Remove a single key from all cache instances
22    Remove { key: String },
23
24    /// Update a key with new value across all cache instances
25    /// This is more efficient than Remove for hot keys as it avoids cache miss
26    Update {
27        key: String,
28        #[serde(with = "serde_bytes_wrapper")]
29        value: Bytes,
30        #[serde(skip_serializing_if = "Option::is_none")]
31        ttl_secs: Option<u64>,
32    },
33
34    /// Remove all keys matching a pattern from all cache instances
35    /// Uses glob-style patterns (e.g., "user:*", "product:123:*")
36    RemovePattern { pattern: String },
37
38    /// Bulk remove multiple keys at once
39    RemoveBulk { keys: Vec<String> },
40}
41
42/// Envelope for invalidation messages supporting instance tracking to avoid self-invalidation echo
43#[derive(Debug, Clone, Serialize, Deserialize)]
44pub struct InvalidationEnvelope {
45    /// Originating instance ID (to prevent self-invalidation echo)
46    #[serde(default, skip_serializing_if = "Option::is_none")]
47    pub origin_id: Option<String>,
48
49    /// The wrapped invalidation message
50    #[serde(flatten)]
51    pub message: InvalidationMessage,
52}
53
54impl InvalidationEnvelope {
55    /// Create a new envelope with optional origin ID
56    #[must_use]
57    pub fn new(message: InvalidationMessage, origin_id: Option<String>) -> Self {
58        Self { origin_id, message }
59    }
60
61    /// Parse JSON payload into an `InvalidationEnvelope`
62    ///
63    /// # Errors
64    /// Returns error if deserialization fails.
65    pub fn parse(json: &str) -> CacheResult<Self> {
66        serde_json::from_str(json).map_err(|e| {
67            crate::error::CacheError::SerializationError(format!(
68                "Failed to deserialize invalidation envelope: {e}"
69            ))
70        })
71    }
72
73    /// Serialize envelope to JSON string
74    ///
75    /// # Errors
76    /// Returns error if serialization fails.
77    pub fn to_json(&self) -> CacheResult<String> {
78        serde_json::to_string(self).map_err(|e| {
79            crate::error::CacheError::SerializationError(format!(
80                "Failed to serialize invalidation envelope: {e}"
81            ))
82        })
83    }
84}
85
86impl InvalidationMessage {
87    /// Create a Remove message
88    pub fn remove(key: impl Into<String>) -> Self {
89        Self::Remove { key: key.into() }
90    }
91
92    /// Create an Update message
93    pub fn update(key: impl Into<String>, value: Bytes, ttl: Option<Duration>) -> Self {
94        Self::Update {
95            key: key.into(),
96            value,
97            ttl_secs: ttl.map(|d| d.as_secs()),
98        }
99    }
100
101    /// Create a `RemovePattern` message
102    pub fn remove_pattern(pattern: impl Into<String>) -> Self {
103        Self::RemovePattern {
104            pattern: pattern.into(),
105        }
106    }
107
108    /// Create a `RemoveBulk` message
109    #[must_use]
110    pub fn remove_bulk(keys: Vec<String>) -> Self {
111        Self::RemoveBulk { keys }
112    }
113
114    /// Serialize to JSON for transmission
115    ///
116    /// # Errors
117    ///
118    /// Returns an error if serialization fails.
119    pub fn to_json(&self) -> CacheResult<String> {
120        serde_json::to_string(self).map_err(|e| {
121            crate::error::CacheError::SerializationError(format!(
122                "Failed to serialize invalidation message: {e}"
123            ))
124        })
125    }
126
127    /// Deserialize from JSON
128    ///
129    /// # Errors
130    ///
131    /// Returns an error if deserialization fails.
132    pub fn from_json(json: &str) -> CacheResult<Self> {
133        serde_json::from_str(json).map_err(|e| {
134            crate::error::CacheError::SerializationError(format!(
135                "Failed to deserialize invalidation message: {e}"
136            ))
137        })
138    }
139
140    /// Get TTL as Duration if present
141    pub fn ttl(&self) -> Option<Duration> {
142        match self {
143            Self::Update { ttl_secs, .. } => ttl_secs.map(Duration::from_secs),
144            _ => None,
145        }
146    }
147}
148
149/// Helper module for Bytes serialization in JSON
150mod serde_bytes_wrapper {
151    use bytes::Bytes;
152    use serde::{Deserialize, Deserializer, Serializer};
153
154    fn parse_hex_digit(b: u8) -> Option<u8> {
155        match b {
156            b'0'..=b'9' => Some(b - b'0'),
157            b'a'..=b'f' => Some(b - b'a' + 10),
158            b'A'..=b'F' => Some(b - b'A' + 10),
159            _ => None,
160        }
161    }
162
163    pub fn serialize<S>(bytes: &Bytes, serializer: S) -> Result<S::Ok, S::Error>
164    where
165        S: Serializer,
166    {
167        let mut hex_str = String::with_capacity(bytes.len() * 2);
168        for &b in bytes.as_ref() {
169            let high = match b >> 4 {
170                val @ 0..=9 => (b'0' + val) as char,
171                val @ 10..=15 => (b'a' + val - 10) as char,
172                _ => '0',
173            };
174            let low = match b & 0xf {
175                val @ 0..=9 => (b'0' + val) as char,
176                val @ 10..=15 => (b'a' + val - 10) as char,
177                _ => '0',
178            };
179            hex_str.push(high);
180            hex_str.push(low);
181        }
182        serializer.serialize_str(&hex_str)
183    }
184
185    pub fn deserialize<'de, D>(deserializer: D) -> Result<Bytes, D::Error>
186    where
187        D: Deserializer<'de>,
188    {
189        let hex_str = String::deserialize(deserializer)?;
190        let bytes_str = hex_str.as_bytes();
191        if bytes_str.len() % 2 != 0 {
192            return Err(serde::de::Error::custom("Odd-length hex string"));
193        }
194        let mut v = Vec::with_capacity(bytes_str.len() / 2);
195        for chunk in bytes_str.chunks_exact(2) {
196            if let [h, l] = chunk {
197                let high = parse_hex_digit(*h)
198                    .ok_or_else(|| serde::de::Error::custom("Invalid hex character"))?;
199                let low = parse_hex_digit(*l)
200                    .ok_or_else(|| serde::de::Error::custom("Invalid hex character"))?;
201                v.push((high << 4) | low);
202            }
203        }
204        Ok(Bytes::from(v))
205    }
206}
207
208/// Configuration for cache invalidation
209#[derive(Debug, Clone)]
210pub struct InvalidationConfig {
211    /// Redis Pub/Sub channel name for invalidation messages
212    pub channel: String,
213
214    /// Whether to automatically broadcast invalidation on writes
215    pub auto_broadcast_on_write: bool,
216
217    /// Whether to also publish invalidation events to Redis Streams for audit
218    pub enable_audit_stream: bool,
219
220    /// Redis Stream name for invalidation audit trail
221    pub audit_stream: String,
222
223    /// Maximum length of audit stream (older entries are trimmed)
224    pub audit_stream_maxlen: Option<usize>,
225}
226
227impl Default for InvalidationConfig {
228    fn default() -> Self {
229        Self {
230            channel: "cache:invalidate".to_string(),
231            auto_broadcast_on_write: false, // Conservative default
232            enable_audit_stream: false,
233            audit_stream: "cache:invalidations".to_string(),
234            audit_stream_maxlen: Some(10000),
235        }
236    }
237}
238
239/// Handle for sending invalidation messages
240pub struct InvalidationPublisher {
241    connection: redis::aio::ConnectionManager,
242    config: InvalidationConfig,
243}
244
245impl InvalidationPublisher {
246    /// Create a new publisher
247    #[must_use]
248    pub fn new(connection: redis::aio::ConnectionManager, config: InvalidationConfig) -> Self {
249        Self { connection, config }
250    }
251
252    /// Publish an invalidation message to all subscribers
253    ///
254    /// # Errors
255    ///
256    /// Returns an error if serialization or publishing fails.
257    pub async fn publish(&self, message: &InvalidationMessage) -> CacheResult<()> {
258        self.publish_with_origin(message, None).await
259    }
260
261    /// Publish an invalidation message with an origin ID to prevent self-invalidation echo
262    ///
263    /// # Errors
264    ///
265    /// Returns an error if serialization or publishing fails.
266    pub async fn publish_with_origin(
267        &self,
268        message: &InvalidationMessage,
269        origin_id: Option<&str>,
270    ) -> CacheResult<()> {
271        let envelope = InvalidationEnvelope {
272            origin_id: origin_id.map(ToString::to_string),
273            message: message.clone(),
274        };
275        let json = envelope.to_json()?;
276
277        // Publish to Pub/Sub channel
278        let mut conn = self.connection.clone();
279        let _: () = conn
280            .publish(&self.config.channel, &json)
281            .await
282            .map_err(|e| {
283                crate::error::CacheError::InvalidationError(format!(
284                    "Failed to publish invalidation message: {e}"
285                ))
286            })?;
287
288        // Optionally publish to audit stream
289        if self.config.enable_audit_stream
290            && let Err(e) = self.publish_to_audit_stream(message, &json).await
291        {
292            // Don't fail the invalidation if audit logging fails
293            warn!("Failed to publish to audit stream: {}", e);
294        }
295
296        Ok(())
297    }
298
299    /// Publish to audit stream for observability
300    async fn publish_to_audit_stream(
301        &self,
302        message: &InvalidationMessage,
303        payload_json: &str,
304    ) -> CacheResult<()> {
305        let timestamp = SystemTime::now()
306            .duration_since(UNIX_EPOCH)
307            .unwrap_or(Duration::ZERO)
308            .as_secs()
309            .to_string();
310
311        // Use &str to avoid unnecessary allocations
312        let (type_str, key_str): (&str, &str);
313        let extra_str: String;
314
315        match message {
316            InvalidationMessage::Remove { key } => {
317                type_str = "remove";
318                key_str = key.as_str();
319                extra_str = String::new();
320            }
321            InvalidationMessage::Update { key, .. } => {
322                type_str = "update";
323                key_str = key.as_str();
324                extra_str = String::new();
325            }
326            InvalidationMessage::RemovePattern { pattern } => {
327                type_str = "remove_pattern";
328                key_str = pattern.as_str();
329                extra_str = String::new();
330            }
331            InvalidationMessage::RemoveBulk { keys } => {
332                type_str = "remove_bulk";
333                key_str = "";
334                extra_str = keys.len().to_string();
335            }
336        }
337
338        let mut fields = vec![
339            ("type", type_str),
340            ("timestamp", timestamp.as_str()),
341            ("payload", payload_json),
342        ];
343
344        if !key_str.is_empty() {
345            fields.push(("key", key_str));
346        }
347        if !extra_str.is_empty() {
348            fields.push(("count", extra_str.as_str()));
349        }
350
351        let mut cmd = redis::cmd("XADD");
352        cmd.arg(&self.config.audit_stream);
353
354        if let Some(maxlen) = self.config.audit_stream_maxlen {
355            cmd.arg("MAXLEN").arg("~").arg(maxlen);
356        }
357
358        cmd.arg("*"); // Auto-generate ID
359
360        for (key, value) in fields {
361            cmd.arg(key).arg(value);
362        }
363
364        let mut conn = self.connection.clone();
365        let _: String = cmd.query_async(&mut conn).await.map_err(|e| {
366            crate::error::CacheError::BackendError(format!("Failed to add to audit stream: {e}"))
367        })?;
368
369        Ok(())
370    }
371}
372
373/// Statistics for invalidation operations
374#[derive(Debug, Default, Clone)]
375pub struct InvalidationStats {
376    /// Number of invalidation messages published
377    pub messages_sent: u64,
378
379    /// Number of invalidation messages received
380    pub messages_received: u64,
381
382    /// Number of Remove operations performed
383    pub removes_received: u64,
384
385    /// Number of Update operations performed
386    pub updates_received: u64,
387
388    /// Number of `RemovePattern` operations performed
389    pub patterns_received: u64,
390
391    /// Number of `RemoveBulk` operations performed
392    pub bulk_removes_received: u64,
393
394    /// Number of failed message processing attempts
395    pub processing_errors: u64,
396}
397
398use std::sync::atomic::{AtomicU64, Ordering};
399
400/// Thread-safe statistics for invalidation operations
401#[derive(Debug, Default)]
402pub struct AtomicInvalidationStats {
403    pub messages_sent: AtomicU64,
404    pub messages_received: AtomicU64,
405    pub removes_received: AtomicU64,
406    pub updates_received: AtomicU64,
407    pub patterns_received: AtomicU64,
408    pub bulk_removes_received: AtomicU64,
409    pub processing_errors: AtomicU64,
410}
411
412impl AtomicInvalidationStats {
413    pub fn snapshot(&self) -> InvalidationStats {
414        InvalidationStats {
415            messages_sent: self.messages_sent.load(Ordering::Relaxed),
416            messages_received: self.messages_received.load(Ordering::Relaxed),
417            removes_received: self.removes_received.load(Ordering::Relaxed),
418            updates_received: self.updates_received.load(Ordering::Relaxed),
419            patterns_received: self.patterns_received.load(Ordering::Relaxed),
420            bulk_removes_received: self.bulk_removes_received.load(Ordering::Relaxed),
421            processing_errors: self.processing_errors.load(Ordering::Relaxed),
422        }
423    }
424}
425
426use std::sync::Arc;
427
428/// Handle for subscribing to invalidation messages
429///
430/// This spawns a background task that listens to Redis Pub/Sub and processes
431/// invalidation messages by calling the provided handler callback.
432pub struct InvalidationSubscriber {
433    /// Redis client for creating Pub/Sub connections
434    client: redis::Client,
435    /// Configuration
436    config: InvalidationConfig,
437    /// Statistics
438    stats: Arc<AtomicInvalidationStats>,
439    /// Shutdown signal sender
440    shutdown_tx: broadcast::Sender<()>,
441    /// Instance ID to filter self-invalidation echo
442    instance_id: Option<String>,
443}
444
445impl InvalidationSubscriber {
446    /// Create a new subscriber
447    ///
448    /// # Arguments
449    /// * `redis_url` - Redis connection URL
450    /// * `config` - Invalidation configuration
451    /// # Errors
452    ///
453    /// Returns an error if Redis client creation fails.
454    pub fn new(redis_url: &str, config: InvalidationConfig) -> CacheResult<Self> {
455        let client = redis::Client::open(redis_url).map_err(|e| {
456            crate::error::CacheError::ConfigError(format!(
457                "Failed to create Redis client for subscriber: {e}"
458            ))
459        })?;
460
461        let (shutdown_tx, _) = broadcast::channel(1);
462
463        Ok(Self {
464            client,
465            config,
466            stats: Arc::new(AtomicInvalidationStats::default()),
467            shutdown_tx,
468            instance_id: None,
469        })
470    }
471
472    /// Set instance ID for filtering out self-invalidation messages
473    #[must_use]
474    pub fn with_instance_id(mut self, instance_id: impl Into<String>) -> Self {
475        self.instance_id = Some(instance_id.into());
476        self
477    }
478
479    /// Get a snapshot of current statistics
480    #[must_use]
481    pub fn stats(&self) -> InvalidationStats {
482        self.stats.snapshot()
483    }
484
485    /// Start the subscriber background task
486    ///
487    /// # Arguments
488    /// * `handler` - Async function to handle each invalidation message
489    ///
490    /// # Returns
491    /// Join handle for the background task
492    pub fn start<F, Fut>(&self, handler: F) -> tokio::task::JoinHandle<()>
493    where
494        F: Fn(InvalidationMessage) -> Fut + Send + Sync + 'static,
495        Fut: std::future::Future<Output = CacheResult<()>> + Send + 'static,
496    {
497        let client = self.client.clone();
498        let channel = self.config.channel.clone();
499        let stats = Arc::clone(&self.stats);
500        let mut shutdown_rx = self.shutdown_tx.subscribe();
501        let instance_id = self.instance_id.clone();
502
503        tokio::spawn(async move {
504            let handler = Arc::new(handler);
505
506            loop {
507                // Check for shutdown signal
508                if shutdown_rx.try_recv().is_ok() {
509                    info!("Invalidation subscriber shutting down...");
510                    break;
511                }
512
513                // Attempt to connect and subscribe
514                match Self::run_subscriber_loop(
515                    &client,
516                    &channel,
517                    Arc::clone(&handler),
518                    Arc::clone(&stats),
519                    instance_id.clone(),
520                    &mut shutdown_rx,
521                )
522                .await
523                {
524                    Ok(()) => {
525                        info!("Invalidation subscriber loop completed normally");
526                        break;
527                    }
528                    Err(e) => {
529                        error!(
530                            "Invalidation subscriber error: {}. Reconnecting in 5s...",
531                            e
532                        );
533                        stats.processing_errors.fetch_add(1, Ordering::Relaxed);
534
535                        // Wait before reconnecting
536                        tokio::select! {
537                            () = tokio::time::sleep(Duration::from_secs(5)) => {},
538                            _ = shutdown_rx.recv() => {
539                                info!("Invalidation subscriber shutting down...");
540                                break;
541                            }
542                        }
543                    }
544                }
545            }
546        })
547    }
548
549    /// Internal subscriber loop
550    async fn run_subscriber_loop<F, Fut>(
551        client: &redis::Client,
552        channel: &str,
553        handler: Arc<F>,
554        stats: Arc<AtomicInvalidationStats>,
555        instance_id: Option<String>,
556        shutdown_rx: &mut broadcast::Receiver<()>,
557    ) -> CacheResult<()>
558    where
559        F: Fn(InvalidationMessage) -> Fut + Send + Sync + 'static,
560        Fut: std::future::Future<Output = CacheResult<()>> + Send + 'static,
561    {
562        let mut pubsub = client.get_async_pubsub().await.map_err(|e| {
563            crate::error::CacheError::BackendError(format!("Failed to get pubsub connection: {e}"))
564        })?;
565
566        // Subscribe to channel
567        pubsub.subscribe(channel).await.map_err(|e| {
568            crate::error::CacheError::InvalidationError(format!(
569                "Failed to subscribe to channel: {e}"
570            ))
571        })?;
572
573        info!("Subscribed to invalidation channel: {}", channel);
574
575        // Get message stream
576        let mut stream = pubsub.on_message();
577
578        loop {
579            // Wait for message or shutdown signal
580            tokio::select! {
581                msg_result = stream.next() => {
582                    match msg_result {
583                        Some(msg) => {
584                            // Get payload
585                            let payload: String = match msg.get_payload() {
586                                Ok(p) => p,
587                                Err(e) => {
588                                    warn!("Failed to get message payload: {}", e);
589                                    stats.processing_errors.fetch_add(1, Ordering::Relaxed);
590                                    continue;
591                                }
592                            };
593
594                            // Deserialize envelope
595                            let envelope = match InvalidationEnvelope::parse(&payload) {
596                                Ok(m) => m,
597                                Err(e) => {
598                                    warn!("Failed to deserialize invalidation message: {}", e);
599                                    stats.processing_errors.fetch_add(1, Ordering::Relaxed);
600                                    continue;
601                                }
602                            };
603
604                            // Filter self-invalidation echo
605                            if let (Some(my_id), Some(sender_id)) = (&instance_id, &envelope.origin_id)
606                                && my_id == sender_id
607                            {
608                                debug!(instance_id = %my_id, "Skipping self-invalidation echo");
609                                continue;
610                            }
611
612                            let invalidation_msg = envelope.message;
613
614                            // Update stats
615                            stats.messages_received.fetch_add(1, Ordering::Relaxed);
616                            match &invalidation_msg {
617                                InvalidationMessage::Remove { .. } => {
618                                    stats.removes_received.fetch_add(1, Ordering::Relaxed);
619                                }
620                                InvalidationMessage::Update { .. } => {
621                                    stats.updates_received.fetch_add(1, Ordering::Relaxed);
622                                }
623                                InvalidationMessage::RemovePattern { .. } => {
624                                    stats.patterns_received.fetch_add(1, Ordering::Relaxed);
625                                }
626                                InvalidationMessage::RemoveBulk { .. } => {
627                                    stats.bulk_removes_received.fetch_add(1, Ordering::Relaxed);
628                                }
629                            }
630
631                            // Call handler
632                            if let Err(e) = handler(invalidation_msg).await {
633                                error!("Invalidation handler error: {}", e);
634                                stats.processing_errors.fetch_add(1, Ordering::Relaxed);
635                            }
636                        }
637                        None => {
638                            // Stream ended
639                            return Err(crate::error::CacheError::InvalidationError("Pub/Sub message stream ended".to_string()));
640                        }
641                    }
642                }
643                _ = shutdown_rx.recv() => {
644                    return Ok(());
645                }
646            }
647        }
648    }
649
650    /// Signal the subscriber to shutdown
651    pub fn shutdown(&self) {
652        let _ = self.shutdown_tx.send(());
653    }
654}
655
656/// Reliable subscriber using Redis Streams and Consumer Groups
657pub struct ReliableStreamSubscriber {
658    redis_url: String,
659    config: InvalidationConfig,
660    stats: Arc<AtomicInvalidationStats>,
661    shutdown_tx: broadcast::Sender<()>,
662    group_name: String,
663    consumer_name: String,
664}
665
666impl ReliableStreamSubscriber {
667    /// Create a new `ReliableStreamSubscriber`
668    ///
669    /// # Errors
670    ///
671    /// Returns an error if the Redis client fails to open.
672    pub fn new(redis_url: &str, config: InvalidationConfig, group_name: &str) -> CacheResult<Self> {
673        let _client = redis::Client::open(redis_url).map_err(|e| {
674            crate::error::CacheError::ConfigError(format!(
675                "Failed to create Redis client for reliable subscriber: {e}"
676            ))
677        })?;
678
679        let (shutdown_tx, _) = broadcast::channel(1);
680        let consumer_name = format!("consumer-{}", Uuid::new_v4());
681
682        Ok(Self {
683            redis_url: redis_url.to_string(),
684            config,
685            stats: Arc::new(AtomicInvalidationStats::default()),
686            shutdown_tx,
687            group_name: group_name.to_string(),
688            consumer_name,
689        })
690    }
691
692    pub fn start<F, Fut>(&self, handler: F) -> tokio::task::JoinHandle<()>
693    where
694        F: Fn(InvalidationMessage) -> Fut + Send + Sync + 'static,
695        Fut: std::future::Future<Output = CacheResult<()>> + Send + 'static,
696    {
697        let stream_key = self.config.channel.clone();
698        let group_name = self.group_name.clone();
699        let consumer_name = self.consumer_name.clone();
700        let handler = Arc::new(handler);
701        let stats = self.stats.clone();
702        let mut shutdown_rx = self.shutdown_tx.subscribe();
703        let redis_url = self.redis_url.clone();
704
705        tokio::spawn(async move {
706            info!(
707                stream = %stream_key,
708                group = %group_name,
709                consumer = %consumer_name,
710                "Starting reliable stream subscriber"
711            );
712
713            // 1. Ensure stream and group exist
714            let redis_backend = crate::redis_streams::RedisStreams::new(&redis_url).await;
715            match redis_backend {
716                Ok(backend) => {
717                    let _ = backend
718                        .stream_create_group(&stream_key, &group_name, "0")
719                        .await;
720
721                    loop {
722                        // Check shutdown before starting loop
723                        if shutdown_rx.try_recv().is_ok() {
724                            break;
725                        }
726
727                        if let Err(e) = Self::run_reliable_loop(
728                            &backend,
729                            &stream_key,
730                            &group_name,
731                            &consumer_name,
732                            handler.clone(),
733                            stats.clone(),
734                            &mut shutdown_rx,
735                        )
736                        .await
737                        {
738                            error!("Reliable subscriber loop error: {}", e);
739
740                            tokio::select! {
741                                () = tokio::time::sleep(Duration::from_secs(5)) => {},
742                                _ = shutdown_rx.recv() => break,
743                            }
744                        } else {
745                            break; // Normal shutdown
746                        }
747                    }
748                }
749                Err(e) => {
750                    error!(
751                        "Failed to initialize Redis Streams backend for reliable subscriber: {}",
752                        e
753                    );
754                }
755            }
756        })
757    }
758
759    async fn run_reliable_loop<F, Fut>(
760        backend: &dyn crate::traits::StreamingBackend,
761        stream_key: &str,
762        group_name: &str,
763        consumer_name: &str,
764        handler: Arc<F>,
765        stats: Arc<AtomicInvalidationStats>,
766        shutdown_rx: &mut broadcast::Receiver<()>,
767    ) -> CacheResult<()>
768    where
769        F: Fn(InvalidationMessage) -> Fut + Send + Sync + 'static,
770        Fut: std::future::Future<Output = CacheResult<()>> + Send + 'static,
771    {
772        loop {
773            tokio::select! {
774                entries_result = backend.stream_read_group(stream_key, group_name, consumer_name, 10, Some(5000)) => {
775                    let entries = entries_result?;
776                    if entries.is_empty() { continue; }
777
778                    let mut processed_ids = Vec::new();
779                    for (id, fields) in entries {
780                        // Find "payload" field or use first field if it looks like JSON
781                        let payload = fields.iter().find(|(k, _)| k == "payload")
782                            .map(|(_, v)| v.as_str())
783                            .or_else(|| fields.first().map(|(_, v)| v.as_str()));
784
785                        if let Some(msg) = payload.and_then(|json| InvalidationEnvelope::parse(json).map(|env| env.message).or_else(|_| InvalidationMessage::from_json(json)).ok()) {
786                            stats.messages_received.fetch_add(1, Ordering::Relaxed);
787                            if let Err(e) = handler(msg).await {
788                                error!("Reliable handler error: {}", e);
789                                stats.processing_errors.fetch_add(1, Ordering::Relaxed);
790                            } else {
791                                processed_ids.push(id);
792                            }
793                        }
794                    }
795
796                    if !processed_ids.is_empty() {
797                        backend.stream_ack(stream_key, group_name, &processed_ids).await?;
798                    }
799                }
800                _ = shutdown_rx.recv() => return Ok(()),
801            }
802        }
803    }
804
805    /// Signal the subscriber to shutdown
806    pub fn shutdown(&self) {
807        let _ = self.shutdown_tx.send(()).unwrap_or(0);
808    }
809}
810
811#[cfg(test)]
812mod tests {
813    use super::*;
814
815    #[test]
816    fn test_invalidation_message_serialization() -> CacheResult<()> {
817        // Test Remove
818        let msg = InvalidationMessage::remove("test_key");
819        let json = msg.to_json()?;
820        let parsed = InvalidationMessage::from_json(&json)?;
821        match parsed {
822            InvalidationMessage::Remove { key } => assert_eq!(key, "test_key"),
823            _ => panic!("Wrong message type"),
824        }
825
826        // Test Update
827        let msg = InvalidationMessage::update(
828            "test_key",
829            Bytes::from("{\"value\": 123}"),
830            Some(Duration::from_secs(3600)),
831        );
832
833        if let InvalidationMessage::Update {
834            key,
835            value,
836            ttl_secs,
837        } = msg
838        {
839            assert_eq!(key, "test_key");
840            assert_eq!(value, Bytes::from("{\"value\": 123}"));
841            assert_eq!(ttl_secs, Some(3600));
842        } else {
843            panic!("Expected Update message");
844        }
845
846        // Test RemovePattern
847        let msg = InvalidationMessage::remove_pattern("user:*");
848        let json = msg.to_json()?;
849        let parsed = InvalidationMessage::from_json(&json)?;
850        match parsed {
851            InvalidationMessage::RemovePattern { pattern } => assert_eq!(pattern, "user:*"),
852            _ => panic!("Wrong message type"),
853        }
854
855        // Test RemoveBulk
856        let msg = InvalidationMessage::remove_bulk(vec!["key1".to_string(), "key2".to_string()]);
857        let json = msg.to_json()?;
858        let parsed = InvalidationMessage::from_json(&json)?;
859        match parsed {
860            InvalidationMessage::RemoveBulk { keys } => assert_eq!(keys, vec!["key1", "key2"]),
861            _ => panic!("Wrong message type"),
862        }
863        Ok(())
864    }
865
866    #[test]
867    fn test_invalidation_config_default() {
868        let config = InvalidationConfig::default();
869        assert_eq!(config.channel, "cache:invalidate");
870        assert!(!config.auto_broadcast_on_write);
871        assert!(!config.enable_audit_stream);
872    }
873
874    #[test]
875    fn test_invalidation_envelope_with_and_without_origin() -> CacheResult<()> {
876        let msg = InvalidationMessage::remove("user:42");
877        let envelope = InvalidationEnvelope::new(msg, Some("node-1".to_string()));
878        let json = envelope.to_json()?;
879
880        let parsed = InvalidationEnvelope::parse(&json)?;
881        assert_eq!(parsed.origin_id, Some("node-1".to_string()));
882        match parsed.message {
883            InvalidationMessage::Remove { key } => assert_eq!(key, "user:42"),
884            _ => panic!("Expected Remove variant"),
885        }
886
887        // Backward compatibility: parsing legacy JSON without origin_id
888        let legacy_json = r#"{"type":"Remove","key":"user:42"}"#;
889        let parsed_legacy = InvalidationEnvelope::parse(legacy_json)?;
890        assert_eq!(parsed_legacy.origin_id, None);
891        match parsed_legacy.message {
892            InvalidationMessage::Remove { key } => assert_eq!(key, "user:42"),
893            _ => panic!("Expected Remove variant"),
894        }
895
896        Ok(())
897    }
898}
multi_tier_cache/invalidation.rs

multi_tier_cache/
invalidation.rs
