Coverage for src/lexigram/notification/delivery/retry.py: 78%

1"""RetryingMailer — wraps a mail backend with exponential backoff retry.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7from lexigram.logging import get_logger 

8from lexigram.notification.delivery.exceptions import PermanentDeliveryFailure 

9from lexigram.result import Err, Ok, Result 

10 

11if TYPE_CHECKING: 

12 from lexigram.contracts.notification.delivery import DeliveryStoreProtocol 

13 

14logger = get_logger(__name__) 

15 

16 

17class RetryingMailer: 

18 """Mail delivery wrapper with exponential backoff and async retry scheduling. 

19 

20 Wraps any mail backend (one returning ``Result[Any, Any]``) with a delivery 

21 store that persists state and schedules deferred retries for transient 

22 failures — instead of sleeping in-process. 

23 

24 Args: 

25 backend: The underlying mail backend with an async ``send()`` method 

26 returning ``Result[Any, Any]``. 

27 store: Delivery state store for tracking retry state. 

28 max_retries: Maximum delivery attempts before permanent failure. 

29 base_delay: Base delay in seconds for exponential backoff. 

30 """ 

31 

32 def __init__( 

33 self, 

34 backend: Any, 

35 store: DeliveryStoreProtocol, 

36 max_retries: int = 3, 

37 base_delay: float = 60.0, 

38 ) -> None: 

39 self._backend = backend 

40 self._store = store 

41 self._max_retries = max_retries 

42 self._base_delay = base_delay 

43 

44 @property 

45 def backend(self) -> Any: 

46 """The wrapped raw mail backend (for flush workers).""" 

47 return self._backend 

48 

49 @property 

50 def store(self) -> DeliveryStoreProtocol: 

51 """The delivery-state store (for flush workers).""" 

52 return self._store # type: ignore[no-any-return] 

53 

54 async def send(self, message: Any) -> Result[str, PermanentDeliveryFailure]: 

55 """Send a message, scheduling a retry on transient failure. 

56 

57 Delegates to the backend. On ``Ok``, marks delivered. On ``Err``, 

58 increments the retry counter: if attempts are below ``max_retries`` 

59 a deferred retry is scheduled via the store; otherwise the delivery 

60 is permanently failed. 

61 

62 Args: 

63 message: Message to send (passed directly to ``backend.send()``). 

64 

65 Returns: 

66 ``Ok(delivery_id)`` on success or when a retry has been scheduled. 

67 ``Err(PermanentDeliveryFailure)`` after max retries exhausted. 

68 """ 

69 delivery_id = await self._store.create_pending(message) 

70 

71 send_result = await self._backend.send(message) 

72 

73 if send_result.is_ok(): 

74 await self._store.mark_delivered(delivery_id) 

75 logger.info("mail_delivered", delivery_id=delivery_id) 

76 return Ok(delivery_id) 

77 

78 attempt = await self._store.increment_retry(delivery_id) 

79 error = send_result.unwrap_err() 

80 

81 if attempt < self._max_retries: 

82 delay = self._base_delay * (2 ** (attempt - 1)) 

83 await self._store.schedule_retry(delivery_id, delay) 

84 logger.warning( 

85 "mail_delivery_retry_scheduled", 

86 delivery_id=delivery_id, 

87 attempt=attempt, 

88 delay_seconds=delay, 

89 error=str(error), 

90 ) 

91 return Ok(delivery_id) 

92 

93 await self._store.mark_failed(delivery_id, reason=str(error)) 

94 logger.error( 

95 "mail_delivery_permanently_failed", 

96 delivery_id=delivery_id, 

97 attempts=attempt, 

98 error=str(error), 

99 ) 

100 return Err(PermanentDeliveryFailure(delivery_id)) 

101 

102 

103__all__ = ["RetryingMailer"]