behavior/protocol/
mod.rs

1//! Neutral typed vocabulary for interpreter-originated event and service lanes.
2//!
3//! Concrete behavior transformations define the closed sum types that add
4//! these lanes. Keeping their values and construction capabilities here avoids
5//! dependencies between otherwise independent transformations.
6
7pub(crate) mod forward;
8
9use std::time::Duration;
10
11use tokio::time::Instant;
12
13use crate::behavior::Address;
14use crate::calculus::UserEvent;
15use crate::{Crash, CreationKind, Exit};
16
17#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
18pub struct TimerId(pub u64);
19
20#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
21pub struct TimerGeneration(pub u64);
22
23impl From<u64> for TimerId {
24    fn from(value: u64) -> Self {
25        Self(value)
26    }
27}
28
29impl From<TimerId> for u64 {
30    fn from(value: TimerId) -> Self {
31        value.0
32    }
33}
34
35impl From<u64> for TimerGeneration {
36    fn from(value: u64) -> Self {
37        Self(value)
38    }
39}
40
41impl From<TimerGeneration> for u64 {
42    fn from(value: TimerGeneration) -> Self {
43        value.0
44    }
45}
46
47#[derive(Debug, Clone, Copy, PartialEq, Eq)]
48pub struct ScheduleAt {
49    pub id: TimerId,
50    pub generation: TimerGeneration,
51    pub at: Instant,
52}
53
54impl ScheduleAt {
55    #[must_use]
56    pub const fn new(id: TimerId, generation: TimerGeneration, at: Instant) -> Self {
57        Self { id, generation, at }
58    }
59}
60
61impl From<(TimerId, TimerGeneration, Instant)> for ScheduleAt {
62    fn from((id, generation, at): (TimerId, TimerGeneration, Instant)) -> Self {
63        Self::new(id, generation, at)
64    }
65}
66
67/// Request scheduling relative to the interpreter's clock.
68///
69/// Constructing this value does not observe a clock. The interpreter resolves
70/// `after` only when it interprets the successful transition that emitted the
71/// request.
72#[derive(Debug, Clone, Copy, PartialEq, Eq)]
73pub struct ScheduleAfter {
74    pub id: TimerId,
75    pub generation: TimerGeneration,
76    pub after: Duration,
77}
78
79impl ScheduleAfter {
80    #[must_use]
81    pub const fn new(id: TimerId, generation: TimerGeneration, after: Duration) -> Self {
82        Self {
83            id,
84            generation,
85            after,
86        }
87    }
88}
89
90impl From<(TimerId, TimerGeneration, Duration)> for ScheduleAfter {
91    fn from((id, generation, after): (TimerId, TimerGeneration, Duration)) -> Self {
92        Self::new(id, generation, after)
93    }
94}
95
96#[derive(Debug, Clone, Copy, PartialEq, Eq)]
97pub struct TimerElapsed {
98    pub id: TimerId,
99    pub generation: TimerGeneration,
100}
101
102impl TimerElapsed {
103    #[must_use]
104    pub const fn new(id: TimerId, generation: TimerGeneration) -> Self {
105        Self { id, generation }
106    }
107}
108
109impl From<(TimerId, TimerGeneration)> for TimerElapsed {
110    fn from((id, generation): (TimerId, TimerGeneration)) -> Self {
111        Self::new(id, generation)
112    }
113}
114
115pub trait TimeEvent: UserEvent {
116    fn time_reached(event: TimerElapsed) -> Option<Self>;
117}
118
119/// Ask the local interpreter to observe the exact peer incarnation selected at
120/// `peer` when this request is interpreted.
121///
122/// [`PeerStopped`] is the pure result protocol. It arrives eventually if a
123/// selected live incarnation later terminates, or may arrive immediately when
124/// the interpreter has authoritative retained termination for the requested
125/// incarnation. Absence from a live-address table is not such authority: an
126/// interpreter that can select neither a live incarnation nor retained
127/// terminal history must return an interpreter error rather than fabricate a
128/// stop result.
129#[derive(Debug, Clone, Copy, PartialEq, Eq)]
130pub struct ObservePeer<A> {
131    pub peer: A,
132}
133
134impl<A> From<A> for ObservePeer<A> {
135    fn from(peer: A) -> Self {
136        Self { peer }
137    }
138}
139
140impl<A> ObservePeer<A> {
141    #[must_use]
142    pub const fn new(peer: A) -> Self {
143        Self { peer }
144    }
145}
146
147/// Ask the local interpreter to cancel this actor's observation of `peer`.
148///
149/// Peer observation is a derived Bombay protocol, not an actor-model
150/// primitive. The address names the same observer-local relationship created
151/// by [`ObservePeer`]; exact-incarnation capture and cancellation belong to the
152/// interpreter. Cancellation does not retract a [`PeerStopped`] event already
153/// admitted to the actor's mailbox, and an interpreter treats a request for a
154/// relationship that is no longer present as inert.
155#[derive(Debug, Clone, Copy, PartialEq, Eq)]
156pub struct UnwatchPeer<A> {
157    pub peer: A,
158}
159
160impl<A> UnwatchPeer<A> {
161    #[must_use]
162    pub const fn new(peer: A) -> Self {
163        Self { peer }
164    }
165}
166
167impl<A> From<A> for UnwatchPeer<A> {
168    fn from(peer: A) -> Self {
169        Self::new(peer)
170    }
171}
172
173#[derive(Debug, Clone, PartialEq, Eq)]
174pub struct PeerStopped<A: Address> {
175    pub peer: A,
176    pub outcome: Result<Exit<A>, Crash>,
177}
178
179impl<A: Address> PeerStopped<A> {
180    #[must_use]
181    pub fn new(peer: A, outcome: Result<Exit<A>, Crash>) -> Self {
182        Self { peer, outcome }
183    }
184}
185
186pub trait PeerEvent: UserEvent {
187    fn peer_stopped(event: PeerStopped<Self::Addr>) -> Option<Self>;
188}
189
190#[derive(Debug, Clone, PartialEq, Eq)]
191pub struct ChildStopped<A: Address> {
192    pub nonce: A::Nonce,
193    pub outcome: Result<Exit<A>, Crash>,
194    pub at: Instant,
195}
196
197impl<A: Address> ChildStopped<A> {
198    #[must_use]
199    pub fn new(nonce: A::Nonce, outcome: Result<Exit<A>, Crash>, at: Instant) -> Self {
200        Self { nonce, outcome, at }
201    }
202}
203
204/// Ask the local interpreter to observe the exact child generation bound at
205/// `nonce`.
206///
207/// Creation is resolved before same-action service sends. If that creation was
208/// rejected, no child exists to observe: the interpreter consumes this request
209/// without installing an observation or emitting [`ChildStopped`]. The
210/// rejection remains observable through [`ObserveCreation`], and a later
211/// creation cannot inherit the consumed observation.
212#[derive(Debug, Clone, Copy, PartialEq, Eq)]
213pub struct ObserveChild<N> {
214    pub nonce: N,
215}
216
217impl<N> ObserveChild<N> {
218    #[must_use]
219    pub const fn new(nonce: N) -> Self {
220        Self { nonce }
221    }
222}
223
224/// A proxy's request for its interpreter to report a worker termination to
225/// the proxy's parent. The interpreter supplies the emitting proxy's child
226/// nonce when constructing [`WorkerStopped`].
227#[derive(Debug, Clone, PartialEq, Eq)]
228pub struct ReportWorkerStopped<A: Address> {
229    pub worker: A::Nonce,
230    pub outcome: Result<Exit<A>, Crash>,
231    pub at: Instant,
232}
233
234impl<A: Address> ReportWorkerStopped<A> {
235    #[must_use]
236    pub fn new(worker: A::Nonce, outcome: Result<Exit<A>, Crash>, at: Instant) -> Self {
237        Self {
238            worker,
239            outcome,
240            at,
241        }
242    }
243}
244
245impl<A: Address> From<ChildStopped<A>> for ReportWorkerStopped<A> {
246    fn from(stopped: ChildStopped<A>) -> Self {
247        Self::new(stopped.nonce, stopped.outcome, stopped.at)
248    }
249}
250
251/// A worker termination reported by a still-live supervised proxy.
252#[derive(Debug, Clone, PartialEq, Eq)]
253pub struct WorkerStopped<A: Address> {
254    pub proxy: A::Nonce,
255    pub worker: A::Nonce,
256    pub outcome: Result<Exit<A>, Crash>,
257    pub at: Instant,
258}
259
260impl<A: Address> WorkerStopped<A> {
261    #[must_use]
262    pub fn new(
263        proxy: A::Nonce,
264        worker: A::Nonce,
265        outcome: Result<Exit<A>, Crash>,
266        at: Instant,
267    ) -> Self {
268        Self {
269            proxy,
270            worker,
271            outcome,
272            at,
273        }
274    }
275}
276
277impl<A: Address> From<(A::Nonce, ReportWorkerStopped<A>)> for WorkerStopped<A> {
278    fn from((proxy, stopped): (A::Nonce, ReportWorkerStopped<A>)) -> Self {
279        Self::new(proxy, stopped.worker, stopped.outcome, stopped.at)
280    }
281}
282
283pub trait ChildEvent: UserEvent {
284    fn child_stopped(event: ChildStopped<Self::Addr>) -> Option<Self>;
285}
286
287pub trait WorkerEvent: UserEvent {
288    fn worker_stopped(event: WorkerStopped<Self::Addr>) -> Option<Self>;
289}
290
291/// Why a staged fresh creation was not committed by an interpreter.
292///
293/// This is a closed semantic classification; interpreter-specific error
294/// values remain at the runtime boundary.
295#[derive(Debug, Clone, Copy, PartialEq, Eq)]
296pub enum CreationRejection {
297    /// The creator-local nonce was already bound, so accepting the request
298    /// would overwrite rather than establish a fresh child.
299    NonceAlreadyBound,
300    /// The fresh child's initialization did not complete successfully.
301    InitializationFailed,
302    /// The interpreter could not allocate, install, or commit the fresh child.
303    EnvironmentFailed,
304}
305
306/// The committed result of one staged [`crate::Create`] request.
307///
308/// `Installed` is emitted only after fresh allocation, successful
309/// initialization, and binding at `nonce`. The replacement provenance is the
310/// provenance supplied by Behavior; an interpreter must never infer it from
311/// address reuse or creation order.
312#[derive(Debug, Clone, Copy, PartialEq, Eq)]
313pub struct CreationResolved<N> {
314    pub nonce: N,
315    pub kind: CreationKind<N>,
316    pub result: Result<(), CreationRejection>,
317}
318
319impl<N> CreationResolved<N> {
320    #[must_use]
321    pub const fn new(
322        nonce: N,
323        kind: CreationKind<N>,
324        result: Result<(), CreationRejection>,
325    ) -> Self {
326        Self {
327            nonce,
328            kind,
329            result,
330        }
331    }
332
333    #[must_use]
334    pub const fn installed(nonce: N, kind: CreationKind<N>) -> Self {
335        Self::new(nonce, kind, Ok(()))
336    }
337
338    /// A successfully committed ordinary birth.
339    #[must_use]
340    pub const fn birth(nonce: N) -> Self {
341        Self::installed(nonce, CreationKind::Birth)
342    }
343
344    /// A successfully committed replacement incarnation.
345    #[must_use]
346    pub const fn replacement_incarnation(nonce: N, replaces: N) -> Self {
347        Self::installed(nonce, CreationKind::ReplacementIncarnation { replaces })
348    }
349
350    #[must_use]
351    pub const fn rejected(nonce: N, kind: CreationKind<N>, rejection: CreationRejection) -> Self {
352        Self::new(nonce, kind, Err(rejection))
353    }
354}
355
356/// Ask the local interpreter to return the committed result of the same-action
357/// creation at `nonce` through the behavior's [`CreationEvent`] lane.
358#[derive(Debug, Clone, Copy, PartialEq, Eq)]
359pub struct ObserveCreation<N> {
360    pub nonce: N,
361}
362
363impl<N> ObserveCreation<N> {
364    #[must_use]
365    pub const fn new(nonce: N) -> Self {
366        Self { nonce }
367    }
368}
369
370pub trait CreationEvent: UserEvent {
371    fn creation_resolved(event: CreationResolved<<Self::Addr as Address>::Nonce>) -> Option<Self>;
372}
373
374/// Ask a proxy's interpreter to report a worker creation result to its parent.
375/// The interpreter supplies the emitting proxy's nonce.
376#[derive(Debug, Clone, Copy, PartialEq, Eq)]
377pub struct ReportWorkerCreationResolved<N> {
378    pub worker: N,
379    pub kind: CreationKind<N>,
380    pub result: Result<(), CreationRejection>,
381}
382
383impl<N> ReportWorkerCreationResolved<N> {
384    #[must_use]
385    pub const fn new(
386        worker: N,
387        kind: CreationKind<N>,
388        result: Result<(), CreationRejection>,
389    ) -> Self {
390        Self {
391            worker,
392            kind,
393            result,
394        }
395    }
396}
397
398impl<N> From<CreationResolved<N>> for ReportWorkerCreationResolved<N> {
399    fn from(resolved: CreationResolved<N>) -> Self {
400        Self::new(resolved.nonce, resolved.kind, resolved.result)
401    }
402}
403
404/// A worker creation result reported by a still-live supervised proxy.
405#[derive(Debug, Clone, Copy, PartialEq, Eq)]
406pub struct WorkerCreationResolved<N> {
407    pub proxy: N,
408    pub worker: N,
409    pub kind: CreationKind<N>,
410    pub result: Result<(), CreationRejection>,
411}
412
413/// Consumer-facing resolution of one explicitly designated replacement.
414///
415/// This is a derived view of [`WorkerCreationResolved`], not another runtime
416/// fact or observation request. `replaced` is the exact prior incarnation
417/// carried by Behavior in [`CreationKind::ReplacementIncarnation`];
418/// `replacement`/`attempt` is the fresh creation nonce. The prior worker's
419/// terminal outcome remains the separate [`WorkerStopped`] fact so creation
420/// resolution cannot duplicate, erase, or reinterpret it.
421#[derive(Debug, Clone, Copy, PartialEq, Eq)]
422pub enum ReplacementResolution<N> {
423    Installed {
424        proxy: N,
425        replaced: N,
426        replacement: N,
427    },
428    Rejected {
429        proxy: N,
430        replaced: N,
431        attempt: N,
432        rejection: CreationRejection,
433    },
434}
435
436impl<N> WorkerCreationResolved<N> {
437    #[must_use]
438    pub const fn new(
439        proxy: N,
440        worker: N,
441        kind: CreationKind<N>,
442        result: Result<(), CreationRejection>,
443    ) -> Self {
444        Self {
445            proxy,
446            worker,
447            kind,
448            result,
449        }
450    }
451
452    /// Project a replacement result without conflating ordinary birth with
453    /// replacement or inferring provenance from nonce arithmetic.
454    #[must_use]
455    pub fn into_replacement(self) -> Option<ReplacementResolution<N>> {
456        let CreationKind::ReplacementIncarnation { replaces } = self.kind else {
457            return None;
458        };
459        Some(match self.result {
460            Ok(()) => ReplacementResolution::Installed {
461                proxy: self.proxy,
462                replaced: replaces,
463                replacement: self.worker,
464            },
465            Err(rejection) => ReplacementResolution::Rejected {
466                proxy: self.proxy,
467                replaced: replaces,
468                attempt: self.worker,
469                rejection,
470            },
471        })
472    }
473}
474
475impl<N> From<(N, ReportWorkerCreationResolved<N>)> for WorkerCreationResolved<N> {
476    fn from((proxy, resolved): (N, ReportWorkerCreationResolved<N>)) -> Self {
477        Self::new(proxy, resolved.worker, resolved.kind, resolved.result)
478    }
479}
480
481pub trait WorkerCreationEvent: UserEvent {
482    fn worker_creation_resolved(
483        event: WorkerCreationResolved<<Self::Addr as Address>::Nonce>,
484    ) -> Option<Self>;
485}
486
487/// A request to finish through one serialized behavior transition.
488#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
489pub struct ShutdownRequested;
490
491pub trait ShutdownEvent: UserEvent {
492    fn shutdown_requested(event: ShutdownRequested) -> Option<Self>;
493}
494
495#[cfg(test)]
496mod tests {
497    use super::*;
498    use crate::MailAddr;
499
500    #[test]
501    fn lifecycle_conversions_preserve_every_semantic_field() {
502        let at = Instant::now();
503        let child = ChildStopped::<MailAddr>::new(3, Err(Crash::Failed), at);
504        let report = ReportWorkerStopped::from(child);
505        let worker = WorkerStopped::from((7, report));
506        assert_eq!(worker.proxy, 7);
507        assert_eq!(worker.worker, 3);
508        assert_eq!(worker.outcome, Err(Crash::Failed));
509        assert_eq!(worker.at, at);
510
511        let creation = CreationResolved::<u64>::rejected(
512            4,
513            CreationKind::replacement_of(3),
514            CreationRejection::EnvironmentFailed,
515        );
516        let report = ReportWorkerCreationResolved::from(creation);
517        let worker = WorkerCreationResolved::from((7, report));
518        assert_eq!(worker.proxy, 7);
519        assert_eq!(worker.worker, 4);
520        assert_eq!(worker.kind, CreationKind::replacement_of(3));
521        assert_eq!(worker.result, Err(CreationRejection::EnvironmentFailed));
522        assert_eq!(
523            worker.into_replacement(),
524            Some(ReplacementResolution::Rejected {
525                proxy: 7,
526                replaced: 3,
527                attempt: 4,
528                rejection: CreationRejection::EnvironmentFailed,
529            })
530        );
531
532        let installed = WorkerCreationResolved::new(7, 5, CreationKind::replacement_of(4), Ok(()));
533        assert_eq!(
534            installed.into_replacement(),
535            Some(ReplacementResolution::Installed {
536                proxy: 7,
537                replaced: 4,
538                replacement: 5,
539            })
540        );
541        assert_eq!(
542            WorkerCreationResolved::new(7, 0, CreationKind::Birth, Ok(())).into_replacement(),
543            None
544        );
545        assert_eq!(
546            WorkerCreationResolved::new(
547                7,
548                0,
549                CreationKind::Birth,
550                Err(CreationRejection::NonceAlreadyBound),
551            )
552            .into_replacement(),
553            None
554        );
555    }
556
557    #[test]
558    fn timer_newtypes_and_requests_have_lossless_construction() {
559        let id = TimerId::from(2);
560        let generation = TimerGeneration::from(5);
561        assert_eq!(u64::from(id), 2);
562        assert_eq!(u64::from(generation), 5);
563        assert_eq!(
564            TimerElapsed::from((id, generation)),
565            TimerElapsed::new(id, generation)
566        );
567    }
568}
behavior/protocol/mod.rs

behavior/protocol/
mod.rs
