1mod config;
141mod error;
142mod events;
143mod layer;
144
145pub use config::{FallbackConfig, FallbackConfigBuilder};
146pub use error::FallbackError;
147pub use events::FallbackEvent;
148pub use layer::FallbackLayer;
149
150use futures::future::BoxFuture;
151use std::sync::Arc;
152use std::task::{Context, Poll};
153use std::time::Instant;
154use tower::Service;
155
156#[cfg(feature = "metrics")]
157use metrics::{counter, describe_counter};
158
159#[cfg(feature = "metrics")]
160use std::sync::Once;
161
162#[cfg(feature = "metrics")]
163static METRICS_INIT: Once = Once::new();
164
165pub type ValueFn<Res> = Arc<dyn Fn() -> Res + Send + Sync>;
167
168pub type FromErrorFn<Res, E> = Arc<dyn Fn(&E) -> Res + Send + Sync>;
170
171pub type FromRequestErrorFn<Req, Res, E> = Arc<dyn Fn(&Req, &E) -> Res + Send + Sync>;
173
174pub type ServiceFn<Req, Res, E> =
176 Arc<dyn Fn(Req) -> BoxFuture<'static, Result<Res, E>> + Send + Sync>;
177
178pub type ExceptionFn<E> = Arc<dyn Fn(E) -> E + Send + Sync>;
180
181pub enum FallbackStrategy<Req, Res, E> {
183 Value(Res),
185
186 ValueFn(ValueFn<Res>),
188
189 FromError(FromErrorFn<Res, E>),
191
192 FromRequestError(FromRequestErrorFn<Req, Res, E>),
194
195 Service(ServiceFn<Req, Res, E>),
198
199 Exception(ExceptionFn<E>),
201}
202
203impl<Req, Res, E> Clone for FallbackStrategy<Req, Res, E>
204where
205 Res: Clone,
206{
207 fn clone(&self) -> Self {
208 match self {
209 Self::Value(v) => Self::Value(v.clone()),
210 Self::ValueFn(f) => Self::ValueFn(Arc::clone(f)),
211 Self::FromError(f) => Self::FromError(Arc::clone(f)),
212 Self::FromRequestError(f) => Self::FromRequestError(Arc::clone(f)),
213 Self::Service(s) => Self::Service(Arc::clone(s)),
214 Self::Exception(f) => Self::Exception(Arc::clone(f)),
215 }
216 }
217}
218
219pub type HandlePredicate<E> = Arc<dyn Fn(&E) -> bool + Send + Sync>;
221
222pub type HandleResponsePredicate<Res> = Arc<dyn Fn(&Res) -> bool + Send + Sync>;
227
228pub struct Fallback<S, Req, Res, E> {
232 inner: S,
233 config: Arc<FallbackConfig<Req, Res, E>>,
234}
235
236impl<S, Req, Res, E> Fallback<S, Req, Res, E> {
237 pub fn new(inner: S, config: Arc<FallbackConfig<Req, Res, E>>) -> Self {
239 #[cfg(feature = "metrics")]
240 METRICS_INIT.call_once(|| {
241 describe_counter!(
242 "fallback_calls_total",
243 "Total number of fallback operations"
244 );
245 });
246
247 Self { inner, config }
248 }
249}
250
251impl<S, Req, Res, E> Clone for Fallback<S, Req, Res, E>
252where
253 S: Clone,
254 Res: Clone,
255{
256 fn clone(&self) -> Self {
257 Self {
258 inner: self.inner.clone(),
259 config: Arc::clone(&self.config),
260 }
261 }
262}
263
264impl<S, Req, Res, E> Service<Req> for Fallback<S, Req, Res, E>
265where
266 S: Service<Req, Response = Res, Error = E> + Clone + Send + 'static,
267 S::Future: Send + 'static,
268 Req: Clone + Send + Sync + 'static,
269 Res: Clone + Send + Sync + 'static,
270 E: Send + Sync + 'static,
271{
272 type Response = Res;
273 type Error = FallbackError<E>;
274 type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
275
276 fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
277 self.inner.poll_ready(cx).map_err(FallbackError::Inner)
278 }
279
280 fn call(&mut self, req: Req) -> Self::Future {
281 let clone = self.inner.clone();
282 let mut service = std::mem::replace(&mut self.inner, clone);
283 let config = Arc::clone(&self.config);
284 let req_clone = req.clone();
285
286 Box::pin(async move {
287 #[cfg(feature = "tracing")]
288 tracing::debug!(fallback = %config.name, "Calling inner service");
289
290 let result = service.call(req).await;
291
292 match result {
293 Ok(response) => {
294 let should_handle_response = config
296 .handle_response_predicate
297 .as_ref()
298 .map(|p| p(&response))
299 .unwrap_or(false);
300
301 if should_handle_response {
302 #[cfg(feature = "tracing")]
303 tracing::debug!(
304 fallback = %config.name,
305 "Response matches predicate, applying fallback"
306 );
307
308 let event = FallbackEvent::FailedAttempt {
310 pattern_name: config.name.clone(),
311 timestamp: Instant::now(),
312 };
313 config.event_listeners.emit(&event);
314
315 match &config.strategy {
317 FallbackStrategy::Value(v) => {
318 #[cfg(feature = "metrics")]
319 counter!(
320 "fallback_calls_total",
321 "fallback" => config.name.clone(),
322 "result" => "applied",
323 "strategy" => "value"
324 )
325 .increment(1);
326
327 let event = FallbackEvent::Applied {
328 pattern_name: config.name.clone(),
329 timestamp: Instant::now(),
330 strategy: "value",
331 };
332 config.event_listeners.emit(&event);
333
334 return Ok(v.clone());
335 }
336
337 FallbackStrategy::ValueFn(f) => {
338 let fallback_response = f();
339
340 #[cfg(feature = "metrics")]
341 counter!(
342 "fallback_calls_total",
343 "fallback" => config.name.clone(),
344 "result" => "applied",
345 "strategy" => "value_fn"
346 )
347 .increment(1);
348
349 let event = FallbackEvent::Applied {
350 pattern_name: config.name.clone(),
351 timestamp: Instant::now(),
352 strategy: "value_fn",
353 };
354 config.event_listeners.emit(&event);
355
356 return Ok(fallback_response);
357 }
358
359 FallbackStrategy::Service(backup) => {
360 #[cfg(feature = "tracing")]
361 tracing::debug!(fallback = %config.name, "Calling backup service (response predicate)");
362
363 match backup(req_clone).await {
364 Ok(backup_response) => {
365 #[cfg(feature = "metrics")]
366 counter!(
367 "fallback_calls_total",
368 "fallback" => config.name.clone(),
369 "result" => "applied",
370 "strategy" => "service"
371 )
372 .increment(1);
373
374 let event = FallbackEvent::Applied {
375 pattern_name: config.name.clone(),
376 timestamp: Instant::now(),
377 strategy: "service",
378 };
379 config.event_listeners.emit(&event);
380
381 return Ok(backup_response);
382 }
383 Err(backup_error) => {
384 #[cfg(feature = "tracing")]
385 tracing::warn!(
386 fallback = %config.name,
387 "Backup service failed (response predicate)"
388 );
389
390 #[cfg(feature = "metrics")]
391 counter!(
392 "fallback_calls_total",
393 "fallback" => config.name.clone(),
394 "result" => "failed",
395 "strategy" => "service"
396 )
397 .increment(1);
398
399 let event = FallbackEvent::Failed {
400 pattern_name: config.name.clone(),
401 timestamp: Instant::now(),
402 };
403 config.event_listeners.emit(&event);
404
405 return Err(FallbackError::FallbackFailed(backup_error));
406 }
407 }
408 }
409
410 _ => {
413 return Ok(response);
414 }
415 }
416 }
417
418 #[cfg(feature = "tracing")]
419 tracing::debug!(fallback = %config.name, "Inner service succeeded");
420
421 #[cfg(feature = "metrics")]
422 counter!(
423 "fallback_calls_total",
424 "fallback" => config.name.clone(),
425 "result" => "success"
426 )
427 .increment(1);
428
429 let event = FallbackEvent::Success {
430 pattern_name: config.name.clone(),
431 timestamp: Instant::now(),
432 };
433 config.event_listeners.emit(&event);
434
435 Ok(response)
436 }
437 Err(error) => {
438 let should_handle = config
440 .handle_predicate
441 .as_ref()
442 .map(|p| p(&error))
443 .unwrap_or(true);
444
445 if !should_handle {
446 #[cfg(feature = "tracing")]
447 tracing::debug!(
448 fallback = %config.name,
449 "Error does not match predicate, skipping fallback"
450 );
451
452 #[cfg(feature = "metrics")]
453 counter!(
454 "fallback_calls_total",
455 "fallback" => config.name.clone(),
456 "result" => "skipped"
457 )
458 .increment(1);
459
460 let event = FallbackEvent::Skipped {
461 pattern_name: config.name.clone(),
462 timestamp: Instant::now(),
463 };
464 config.event_listeners.emit(&event);
465
466 return Err(FallbackError::Inner(error));
467 }
468
469 #[cfg(feature = "tracing")]
470 tracing::debug!(fallback = %config.name, "Inner service failed, applying fallback");
471
472 let event = FallbackEvent::FailedAttempt {
474 pattern_name: config.name.clone(),
475 timestamp: Instant::now(),
476 };
477 config.event_listeners.emit(&event);
478
479 match &config.strategy {
481 FallbackStrategy::Value(v) => {
482 #[cfg(feature = "metrics")]
483 counter!(
484 "fallback_calls_total",
485 "fallback" => config.name.clone(),
486 "result" => "applied",
487 "strategy" => "value"
488 )
489 .increment(1);
490
491 let event = FallbackEvent::Applied {
492 pattern_name: config.name.clone(),
493 timestamp: Instant::now(),
494 strategy: "value",
495 };
496 config.event_listeners.emit(&event);
497
498 Ok(v.clone())
499 }
500
501 FallbackStrategy::ValueFn(f) => {
502 let response = f();
503
504 #[cfg(feature = "metrics")]
505 counter!(
506 "fallback_calls_total",
507 "fallback" => config.name.clone(),
508 "result" => "applied",
509 "strategy" => "value_fn"
510 )
511 .increment(1);
512
513 let event = FallbackEvent::Applied {
514 pattern_name: config.name.clone(),
515 timestamp: Instant::now(),
516 strategy: "value_fn",
517 };
518 config.event_listeners.emit(&event);
519
520 Ok(response)
521 }
522
523 FallbackStrategy::FromError(f) => {
524 let response = f(&error);
525
526 #[cfg(feature = "metrics")]
527 counter!(
528 "fallback_calls_total",
529 "fallback" => config.name.clone(),
530 "result" => "applied",
531 "strategy" => "from_error"
532 )
533 .increment(1);
534
535 let event = FallbackEvent::Applied {
536 pattern_name: config.name.clone(),
537 timestamp: Instant::now(),
538 strategy: "from_error",
539 };
540 config.event_listeners.emit(&event);
541
542 Ok(response)
543 }
544
545 FallbackStrategy::FromRequestError(f) => {
546 let response = f(&req_clone, &error);
547
548 #[cfg(feature = "metrics")]
549 counter!(
550 "fallback_calls_total",
551 "fallback" => config.name.clone(),
552 "result" => "applied",
553 "strategy" => "from_request_error"
554 )
555 .increment(1);
556
557 let event = FallbackEvent::Applied {
558 pattern_name: config.name.clone(),
559 timestamp: Instant::now(),
560 strategy: "from_request_error",
561 };
562 config.event_listeners.emit(&event);
563
564 Ok(response)
565 }
566
567 FallbackStrategy::Service(backup) => {
568 #[cfg(feature = "tracing")]
569 tracing::debug!(fallback = %config.name, "Calling backup service");
570
571 match backup(req_clone).await {
572 Ok(response) => {
573 #[cfg(feature = "metrics")]
574 counter!(
575 "fallback_calls_total",
576 "fallback" => config.name.clone(),
577 "result" => "applied",
578 "strategy" => "service"
579 )
580 .increment(1);
581
582 let event = FallbackEvent::Applied {
583 pattern_name: config.name.clone(),
584 timestamp: Instant::now(),
585 strategy: "service",
586 };
587 config.event_listeners.emit(&event);
588
589 Ok(response)
590 }
591 Err(backup_error) => {
592 #[cfg(feature = "tracing")]
593 tracing::warn!(
594 fallback = %config.name,
595 "Backup service also failed"
596 );
597
598 #[cfg(feature = "metrics")]
599 counter!(
600 "fallback_calls_total",
601 "fallback" => config.name.clone(),
602 "result" => "failed",
603 "strategy" => "service"
604 )
605 .increment(1);
606
607 let event = FallbackEvent::Failed {
608 pattern_name: config.name.clone(),
609 timestamp: Instant::now(),
610 };
611 config.event_listeners.emit(&event);
612
613 Err(FallbackError::FallbackFailed(backup_error))
614 }
615 }
616 }
617
618 FallbackStrategy::Exception(transform) => {
619 let transformed = transform(error);
620
621 #[cfg(feature = "metrics")]
622 counter!(
623 "fallback_calls_total",
624 "fallback" => config.name.clone(),
625 "result" => "transformed",
626 "strategy" => "exception"
627 )
628 .increment(1);
629
630 let event = FallbackEvent::Applied {
631 pattern_name: config.name.clone(),
632 timestamp: Instant::now(),
633 strategy: "exception",
634 };
635 config.event_listeners.emit(&event);
636
637 Err(FallbackError::Inner(transformed))
638 }
639 }
640 }
641 }
642 })
643 }
644}
645
646#[cfg(test)]
647mod tests {
648 use super::*;
649 use std::sync::atomic::{AtomicUsize, Ordering};
650 use tower::{service_fn, Layer, ServiceExt};
651 use tower_resilience_core::ResilienceEvent;
652
653 #[derive(Debug, Clone)]
654 struct TestError {
655 message: String,
656 retryable: bool,
657 }
658
659 impl TestError {
660 fn new(message: &str) -> Self {
661 Self {
662 message: message.to_string(),
663 retryable: true,
664 }
665 }
666
667 fn non_retryable(message: &str) -> Self {
668 Self {
669 message: message.to_string(),
670 retryable: false,
671 }
672 }
673 }
674
675 #[tokio::test]
676 async fn success_no_fallback() {
677 let service =
678 service_fn(
679 |req: String| async move { Ok::<_, TestError>(format!("response: {}", req)) },
680 );
681
682 let layer = FallbackLayer::<String, String, TestError>::value("fallback".to_string());
683 let mut service = layer.layer(service);
684
685 let response = service
686 .ready()
687 .await
688 .unwrap()
689 .call("test".to_string())
690 .await
691 .unwrap();
692
693 assert_eq!(response, "response: test");
694 }
695
696 #[tokio::test]
697 async fn failure_triggers_value_fallback() {
698 let service =
699 service_fn(|_req: String| async move { Err::<String, _>(TestError::new("failed")) });
700
701 let layer = FallbackLayer::<String, String, TestError>::value("fallback".to_string());
702 let mut service = layer.layer(service);
703
704 let response = service
705 .ready()
706 .await
707 .unwrap()
708 .call("test".to_string())
709 .await
710 .unwrap();
711
712 assert_eq!(response, "fallback");
713 }
714
715 #[tokio::test]
716 async fn failure_triggers_from_error_fallback() {
717 let service = service_fn(|_req: String| async move {
718 Err::<String, _>(TestError::new("something went wrong"))
719 });
720
721 let layer = FallbackLayer::<String, String, TestError>::from_error(|e: &TestError| {
722 format!("Error: {}", e.message)
723 });
724 let mut service = layer.layer(service);
725
726 let response = service
727 .ready()
728 .await
729 .unwrap()
730 .call("test".to_string())
731 .await
732 .unwrap();
733
734 assert_eq!(response, "Error: something went wrong");
735 }
736
737 #[tokio::test]
738 async fn failure_triggers_from_request_error_fallback() {
739 let service =
740 service_fn(|_req: String| async move { Err::<String, _>(TestError::new("failed")) });
741
742 let layer = FallbackLayer::<String, String, TestError>::from_request_error(
743 |req: &String, _e: &TestError| format!("fallback for: {}", req),
744 );
745 let mut service = layer.layer(service);
746
747 let response = service
748 .ready()
749 .await
750 .unwrap()
751 .call("my-request".to_string())
752 .await
753 .unwrap();
754
755 assert_eq!(response, "fallback for: my-request");
756 }
757
758 #[tokio::test]
759 async fn predicate_skips_non_matching_errors() {
760 let service = service_fn(|_req: String| async move {
761 Err::<String, _>(TestError::non_retryable("permanent failure"))
762 });
763
764 let layer = FallbackLayer::builder()
765 .value("fallback".to_string())
766 .handle(|e: &TestError| e.retryable) .build();
768 let mut service = layer.layer(service);
769
770 let result = service
771 .ready()
772 .await
773 .unwrap()
774 .call("test".to_string())
775 .await;
776
777 assert!(matches!(result, Err(FallbackError::Inner(_))));
779 }
780
781 #[tokio::test]
782 async fn backup_service_fallback() {
783 let call_count = Arc::new(AtomicUsize::new(0));
784 let cc = Arc::clone(&call_count);
785
786 let primary = service_fn(move |_req: String| {
787 let cc = Arc::clone(&cc);
788 async move {
789 cc.fetch_add(1, Ordering::SeqCst);
790 Err::<String, _>(TestError::new("primary failed"))
791 }
792 });
793
794 let backup_calls = Arc::new(AtomicUsize::new(0));
795 let bc = Arc::clone(&backup_calls);
796
797 let layer = FallbackLayer::<String, String, TestError>::service(move |req: String| {
798 let bc = Arc::clone(&bc);
799 async move {
800 bc.fetch_add(1, Ordering::SeqCst);
801 Ok::<_, TestError>(format!("backup: {}", req))
802 }
803 });
804 let mut service = layer.layer(primary);
805
806 let response = service
807 .ready()
808 .await
809 .unwrap()
810 .call("test".to_string())
811 .await
812 .unwrap();
813
814 assert_eq!(response, "backup: test");
815 assert_eq!(call_count.load(Ordering::SeqCst), 1);
816 assert_eq!(backup_calls.load(Ordering::SeqCst), 1);
817 }
818
819 #[tokio::test]
820 async fn backup_service_also_fails() {
821 let primary =
822 service_fn(
823 |_req: String| async move { Err::<String, _>(TestError::new("primary failed")) },
824 );
825
826 let layer =
827 FallbackLayer::<String, String, TestError>::service(|_req: String| async move {
828 Err::<String, _>(TestError::new("backup also failed"))
829 });
830 let mut service = layer.layer(primary);
831
832 let result = service
833 .ready()
834 .await
835 .unwrap()
836 .call("test".to_string())
837 .await;
838
839 assert!(matches!(result, Err(FallbackError::FallbackFailed(_))));
840 }
841
842 #[tokio::test]
843 async fn exception_transforms_error() {
844 let service =
845 service_fn(
846 |_req: String| async move { Err::<String, _>(TestError::new("original error")) },
847 );
848
849 let layer = FallbackLayer::<String, String, TestError>::exception(|_e: TestError| {
850 TestError::new("transformed error")
851 });
852 let mut service = layer.layer(service);
853
854 let result = service
855 .ready()
856 .await
857 .unwrap()
858 .call("test".to_string())
859 .await;
860
861 match result {
862 Err(FallbackError::Inner(e)) => {
863 assert_eq!(e.message, "transformed error");
864 }
865 _ => panic!("expected transformed error"),
866 }
867 }
868
869 #[tokio::test]
870 async fn event_listeners_called() {
871 use std::sync::Mutex;
872
873 let events: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
874 let events_clone = Arc::clone(&events);
875
876 let service =
877 service_fn(|_req: String| async move { Err::<String, _>(TestError::new("failed")) });
878
879 let layer = FallbackLayer::builder()
880 .name("test-fallback")
881 .value("fallback".to_string())
882 .on_event(move |event: &FallbackEvent| {
883 events_clone
884 .lock()
885 .unwrap()
886 .push(event.event_type().to_string());
887 })
888 .build();
889 let mut service = layer.layer(service);
890
891 let _ = service
892 .ready()
893 .await
894 .unwrap()
895 .call("test".to_string())
896 .await;
897
898 let recorded = events.lock().unwrap();
899 assert!(recorded.contains(&"failed_attempt".to_string()));
900 assert!(recorded.contains(&"applied".to_string()));
901 }
902
903 #[derive(Debug)]
908 struct NonCloneError(String);
909
910 #[tokio::test]
911 async fn works_with_non_clone_error() {
912 let service = service_fn(|_req: String| async move {
914 Err::<String, _>(NonCloneError("boom".to_string()))
915 });
916 let layer =
917 FallbackLayer::<String, String, NonCloneError>::from_error(|e: &NonCloneError| {
918 format!("recovered: {}", e.0)
919 });
920 let mut service = layer.layer(service);
921 let response = service
922 .ready()
923 .await
924 .unwrap()
925 .call("test".to_string())
926 .await
927 .unwrap();
928 assert_eq!(response, "recovered: boom");
929
930 let service = service_fn(|_req: String| async move {
932 Err::<String, _>(NonCloneError("permanent".to_string()))
933 });
934 let layer = FallbackLayer::builder()
935 .value("unused".to_string())
936 .handle(|_e: &NonCloneError| false) .build();
938 let mut service = layer.layer(service);
939 let result = service
940 .ready()
941 .await
942 .unwrap()
943 .call("test".to_string())
944 .await;
945 match result {
946 Err(FallbackError::Inner(e)) => assert_eq!(e.0, "permanent"),
947 _ => panic!("expected original error to propagate"),
948 }
949 }
950}