behavior/timing/
receive_timeout.rs

1//! Pure receive-inactivity composition. Relative scheduling is a request to
2//! the emitting actor's interpreter and never observes a clock here.
3
4use std::time::Duration;
5
6use super::domain::TimerLease;
7use super::event::TimedEvent;
8use crate::Step;
9use crate::behavior::{
10    Actions, Address, Behavior, BirthMode, SendAlgebra, ServiceSends, UserEvent,
11};
12use crate::protocol::{ScheduleAfter, TimeEvent, TimerId};
13use crate::{Inner, Own, SendInput};
14
15pub type ReceiveTimeoutEvent<E> = TimedEvent<E>;
16
17/// A controlled receive-timeout failure.
18#[derive(Debug, Clone, PartialEq, Eq)]
19pub enum ReceiveTimeoutError<E> {
20    /// The inner fold or timeout reaction failed.
21    Inner(E),
22    /// Advancing the timer generation would make a stale delivery live again.
23    ///
24    /// This is detected after the successful continuing inner user fold: the
25    /// inner state mutation has occurred, but its returned sends and creations
26    /// are not emitted because the composed transition fails. Bombay behavior
27    /// folds are not transactional and wrappers cannot roll back inner state.
28    GenerationExhausted,
29}
30
31pub type ReceiveTimeoutReaction<B> = fn(
32    &mut B,
33) -> Result<
34    Actions<
35        <B as Behavior>::Addr,
36        <B as Behavior>::Ph,
37        <B as Behavior>::Sends,
38        <B as Behavior>::Birth,
39    >,
40    <B as Behavior>::Error,
41>;
42
43/// Named effect lanes added by [`ReceiveTimeout`].
44pub struct ReceiveTimeoutSends<Sends> {
45    pub behavior: Sends,
46    pub schedules: ServiceSends<ScheduleAfter>,
47}
48
49impl<Sends: SendAlgebra> SendAlgebra for ReceiveTimeoutSends<Sends> {
50    fn empty() -> Self {
51        Self {
52            behavior: Sends::empty(),
53            schedules: ServiceSends::empty(),
54        }
55    }
56
57    fn append(&mut self, other: Self) {
58        self.behavior.append(other.behavior);
59        self.schedules.append(other.schedules);
60    }
61}
62
63impl<Sends> SendInput<ScheduleAfter, Own> for ReceiveTimeoutSends<Sends> {
64    fn emit(&mut self, input: ScheduleAfter) {
65        self.schedules.send(input);
66    }
67}
68
69impl<Sends, Input, Path> SendInput<Input, Inner<Path>> for ReceiveTimeoutSends<Sends>
70where
71    Sends: SendInput<Input, Path>,
72{
73    fn emit(&mut self, input: Input) {
74        <Sends as SendInput<Input, Path>>::emit(&mut self.behavior, input);
75    }
76}
77
78pub type ReceiveTimeoutActions<B> = Actions<
79    <B as Behavior>::Addr,
80    <B as Behavior>::Ph,
81    ReceiveTimeoutSends<<B as Behavior>::Sends>,
82    <B as Behavior>::Birth,
83>;
84
85/// A pure one-notification-per-idle-period receive timeout.
86///
87/// Only successful user communications are activity. Timer, peer, child,
88/// worker, and shutdown service events compose through this wrapper but never
89/// rearm it. A matching timeout consumes the live generation before invoking
90/// the reaction; if that reaction continues, the timeout remains unarmed until
91/// another successful continuing user communication.
92pub struct ReceiveTimeout<B: Behavior> {
93    inner: B,
94    id: TimerId,
95    after: Duration,
96    timer: TimerLease,
97    on_elapsed: ReceiveTimeoutReaction<B>,
98}
99
100impl<B: Behavior> ReceiveTimeout<B> {
101    #[must_use]
102    pub fn new(
103        inner: B,
104        id: TimerId,
105        after: Duration,
106        on_elapsed: ReceiveTimeoutReaction<B>,
107    ) -> Self {
108        Self {
109            inner,
110            id,
111            after,
112            timer: TimerLease::new(),
113            on_elapsed,
114        }
115    }
116
117    #[must_use]
118    pub fn inner(&self) -> &B {
119        &self.inner
120    }
121
122    fn schedule(&mut self) -> Result<ServiceSends<ScheduleAfter>, ReceiveTimeoutError<B::Error>> {
123        let generation = self
124            .timer
125            .arm()
126            .map_err(|_| ReceiveTimeoutError::GenerationExhausted)?;
127        Ok(ServiceSends::one(ScheduleAfter::new(
128            self.id, generation, self.after,
129        )))
130    }
131
132    fn wrap(
133        actions: Actions<B::Addr, B::Ph, B::Sends, B::Birth>,
134        own: ServiceSends<ScheduleAfter>,
135    ) -> ReceiveTimeoutActions<B> {
136        actions.map_sends(|behavior| ReceiveTimeoutSends {
137            behavior,
138            schedules: own,
139        })
140    }
141
142    fn terminal(actions: &Actions<B::Addr, B::Ph, B::Sends, B::Birth>) -> bool {
143        matches!(actions.become_, Step::Stop(_))
144    }
145}
146
147impl<B, A, Ph, Sends, Br> Behavior for ReceiveTimeout<B>
148where
149    A: Address,
150    Sends: SendAlgebra,
151    Br: BirthMode,
152    B: Behavior<Addr = A, Ph = Ph, Sends = Sends, Birth = Br>,
153    B::Event: TimeEvent,
154{
155    type Addr = A;
156    type Msg = B::Msg;
157    type Event = ReceiveTimeoutEvent<B::Event>;
158    type Sends = ReceiveTimeoutSends<Sends>;
159    type Ph = Ph;
160    type Error = ReceiveTimeoutError<B::Error>;
161    type Birth = Br;
162
163    fn init(&mut self) -> Result<ReceiveTimeoutActions<B>, Self::Error> {
164        let actions = self.inner.init().map_err(ReceiveTimeoutError::Inner)?;
165        let own = if Self::terminal(&actions) {
166            self.timer.disarm();
167            ServiceSends::empty()
168        } else {
169            self.schedule()?
170        };
171        Ok(Self::wrap(actions, own))
172    }
173
174    fn transition(&mut self, event: Self::Event) -> Result<ReceiveTimeoutActions<B>, Self::Error> {
175        match event {
176            ReceiveTimeoutEvent::Elapsed(elapsed)
177                if elapsed.id == self.id && self.timer.accept(elapsed.generation) =>
178            {
179                let actions =
180                    (self.on_elapsed)(&mut self.inner).map_err(ReceiveTimeoutError::Inner)?;
181                Ok(Self::wrap(actions, ServiceSends::empty()))
182            }
183            ReceiveTimeoutEvent::Elapsed(elapsed) if elapsed.id == self.id => Ok(Actions::cont()),
184            ReceiveTimeoutEvent::Elapsed(elapsed) => {
185                let Some(inner) = B::Event::time_reached(elapsed) else {
186                    return Ok(Actions::cont());
187                };
188                let actions = self
189                    .inner
190                    .transition(inner)
191                    .map_err(ReceiveTimeoutError::Inner)?;
192                if Self::terminal(&actions) {
193                    self.timer.disarm();
194                }
195                Ok(Self::wrap(actions, ServiceSends::empty()))
196            }
197            ReceiveTimeoutEvent::Inner(event) => match event.into_user() {
198                Ok(user) => {
199                    let event = B::Event::user(user.from, user.message);
200                    let actions = self
201                        .inner
202                        .transition(event)
203                        .map_err(ReceiveTimeoutError::Inner)?;
204                    let own = if Self::terminal(&actions) {
205                        self.timer.disarm();
206                        ServiceSends::empty()
207                    } else {
208                        self.schedule()?
209                    };
210                    Ok(Self::wrap(actions, own))
211                }
212                Err(service) => {
213                    let actions = self
214                        .inner
215                        .transition(service)
216                        .map_err(ReceiveTimeoutError::Inner)?;
217                    if Self::terminal(&actions) {
218                        self.timer.disarm();
219                    }
220                    Ok(Self::wrap(actions, ServiceSends::empty()))
221                }
222            },
223        }
224    }
225}
226
227#[cfg(test)]
228mod tests {
229    use super::*;
230    use crate::{Acted, Handler, MailAddr, Never, NoBirths, Pure, TimerGeneration, User};
231
232    struct Count(u8);
233
234    impl Handler for Count {
235        type Addr = MailAddr;
236        type Msg = ();
237
238        fn receive(
239            &mut self,
240            _from: MailAddr,
241            (): (),
242        ) -> Acted<MailAddr, Never, Vec<Never>, NoBirths, Never> {
243            self.0 += 1;
244            Ok(Actions::cont())
245        }
246    }
247
248    type Inner = Pure<Count>;
249
250    #[allow(
251        clippy::unnecessary_wraps,
252        reason = "the reaction fixture must implement the fallible reaction signature"
253    )]
254    fn elapsed(_inner: &mut Inner) -> Acted<MailAddr, Never, Vec<Never>, NoBirths, Never> {
255        Ok(Actions::cont())
256    }
257
258    #[tokio::test]
259    async fn exhaustion_follows_the_successful_inner_fold_without_emitting_its_actions() {
260        let mut timeout = ReceiveTimeout::new(
261            Pure::new(Count(0)),
262            TimerId(0),
263            Duration::from_secs(1),
264            elapsed,
265        );
266        timeout.init().unwrap();
267        timeout.timer = TimerLease::idle(TimerGeneration(u64::MAX));
268
269        let result = timeout.transition(ReceiveTimeoutEvent::Inner(User::user(MailAddr(1), ())));
270
271        assert!(matches!(
272            result,
273            Err(ReceiveTimeoutError::GenerationExhausted)
274        ));
275        assert_eq!(timeout.inner().state().0, 1);
276        assert_eq!(timeout.timer.live(), None);
277    }
278}
behavior/timing/receive_timeout.rs

behavior/timing/
receive_timeout.rs
