liminal/conversation/
types.rs

1use std::sync::Arc;
2use std::time::{Duration, Instant};
3
4use beamr::process::ExitReason;
5use beamr::process::registry::ProcessHandle;
6
7use crate::channel::ChannelMode;
8use crate::envelope::Envelope;
9use crate::error::LiminalError;
10use crate::tracing::{ConversationSpan, FinishedSpan, TraceContext};
11
12#[derive(Debug)]
13pub struct Conversation {
14    span: ConversationSpan,
15}
16
17impl Conversation {
18    #[must_use]
19    pub fn start(conversation_id: impl Into<String>) -> Self {
20        Self {
21            span: ConversationSpan::root(conversation_id),
22        }
23    }
24
25    #[must_use]
26    pub fn spawn_child(&self, conversation_id: impl Into<String>) -> Self {
27        Self {
28            span: self.span.child(conversation_id),
29        }
30    }
31
32    #[must_use]
33    pub const fn message<Payload>(&self, payload: Payload) -> ConversationMessage<Payload> {
34        ConversationMessage::new(payload, self.span.message_context())
35    }
36
37    #[must_use]
38    pub fn name(&self) -> &str {
39        self.span.name()
40    }
41
42    #[must_use]
43    pub const fn trace_context(&self) -> TraceContext {
44        self.span.context()
45    }
46
47    #[must_use]
48    pub const fn parent_trace_context(&self) -> Option<TraceContext> {
49        self.span.parent()
50    }
51
52    #[must_use]
53    pub fn finish(self) -> FinishedSpan {
54        self.span.finish()
55    }
56}
57
58#[derive(Clone, Debug, PartialEq, Eq)]
59pub struct ConversationMessage<Payload> {
60    payload: Payload,
61    trace_context: TraceContext,
62}
63
64impl<Payload> ConversationMessage<Payload> {
65    const fn new(payload: Payload, trace_context: TraceContext) -> Self {
66        Self {
67            payload,
68            trace_context,
69        }
70    }
71
72    #[must_use]
73    pub const fn trace_context(&self) -> TraceContext {
74        self.trace_context
75    }
76
77    #[must_use]
78    pub const fn payload(&self) -> &Payload {
79        &self.payload
80    }
81
82    #[must_use]
83    pub fn into_payload(self) -> Payload {
84        self.payload
85    }
86
87    #[must_use]
88    pub fn map<NextPayload>(
89        self,
90        map_payload: impl FnOnce(Payload) -> NextPayload,
91    ) -> ConversationMessage<NextPayload> {
92        ConversationMessage {
93            payload: map_payload(self.payload),
94            trace_context: self.trace_context,
95        }
96    }
97}
98
99/// PID of a beamr process participating in a conversation.
100#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
101pub struct ParticipantPid(u64);
102
103impl ParticipantPid {
104    /// Creates a participant PID wrapper from a raw beamr PID.
105    #[must_use]
106    pub const fn new(pid: u64) -> Self {
107        Self(pid)
108    }
109
110    /// Returns the raw beamr PID.
111    #[must_use]
112    pub const fn get(self) -> u64 {
113        self.0
114    }
115}
116
117impl From<u64> for ParticipantPid {
118    fn from(pid: u64) -> Self {
119        Self::new(pid)
120    }
121}
122
123impl From<ProcessHandle> for ParticipantPid {
124    fn from(handle: ProcessHandle) -> Self {
125        Self::new(handle.pid())
126    }
127}
128
129/// Policy applied when a linked participant exits unexpectedly.
130#[derive(Clone, Copy, Debug, PartialEq, Eq)]
131pub enum CrashPolicy {
132    /// Fail the conversation immediately.
133    Fail,
134    /// Record that routing should select another participant in a later brief.
135    RouteToNext,
136    /// Record that compensation should run in a later brief.
137    Compensate,
138}
139
140/// Required configuration for a conversation actor.
141#[derive(Clone, Debug, PartialEq, Eq)]
142pub struct ConversationConfig {
143    /// Participant process identifiers linked to the conversation actor.
144    pub participants: Vec<ParticipantPid>,
145    /// Optional in-memory deadline for the conversation.
146    pub timeout: Option<Duration>,
147    /// Durability mode marker; durable persistence is implemented elsewhere.
148    pub mode: ChannelMode,
149    /// Participant-crash policy. This has no default and must be provided.
150    pub on_crash: CrashPolicy,
151}
152
153impl ConversationConfig {
154    /// Creates conversation configuration from all required fields.
155    #[must_use]
156    pub const fn new(
157        participants: Vec<ParticipantPid>,
158        timeout: Option<Duration>,
159        mode: ChannelMode,
160        on_crash: CrashPolicy,
161    ) -> Self {
162        Self {
163            participants,
164            timeout,
165            mode,
166            on_crash,
167        }
168    }
169}
170
171/// Lifecycle phase of a conversation actor.
172#[derive(Clone, Copy, Debug, PartialEq, Eq)]
173pub enum ConversationPhase {
174    /// Actor exists but no exchange has started.
175    Created,
176    /// Messages are being exchanged.
177    Active,
178    /// Normal close has begun.
179    Completing,
180    /// Normal close finished.
181    Closed,
182    /// The conversation failed.
183    Failed,
184}
185
186/// Liveness recorded for one linked participant.
187#[derive(Clone, Copy, Debug, PartialEq, Eq)]
188pub enum ParticipantHealth {
189    /// The participant has not emitted an EXIT signal.
190    Alive,
191    /// The participant emitted an EXIT signal.
192    Dead,
193}
194
195/// Participant liveness snapshot.
196#[derive(Clone, Copy, Debug, PartialEq, Eq)]
197pub struct ParticipantStatus {
198    /// Participant PID.
199    pub participant: ParticipantPid,
200    /// Last known participant liveness.
201    pub health: ParticipantHealth,
202    /// Instant the participant's EXIT signal was observed, set when the
203    /// participant is marked dead. Replayed to a late exit-notifier registrant
204    /// so a crash that lands before registration is not lost.
205    pub exited_at: Option<Instant>,
206    /// EXIT reason carried by the participant's trapped exit signal, set when
207    /// the participant is marked dead. `None` while alive, or when the death
208    /// was discovered without a signal (boot pruning a pid that died while no
209    /// actor was linked to it).
210    pub exit_reason: Option<ExitReason>,
211}
212
213impl ParticipantStatus {
214    /// Creates an alive participant status.
215    #[must_use]
216    pub const fn alive(participant: ParticipantPid) -> Self {
217        Self {
218            participant,
219            health: ParticipantHealth::Alive,
220            exited_at: None,
221            exit_reason: None,
222        }
223    }
224
225    /// Marks this participant dead, recording the instant its EXIT was observed
226    /// and the reason its exit signal carried (`None` when the death was
227    /// discovered without a signal).
228    pub const fn mark_dead_at(&mut self, at: Instant, reason: Option<ExitReason>) {
229        self.health = ParticipantHealth::Dead;
230        self.exited_at = Some(at);
231        self.exit_reason = reason;
232    }
233}
234
235/// Opaque conversation context accumulated by the actor.
236#[derive(Clone, Debug, PartialEq, Eq)]
237pub enum ConversationContextEntry {
238    /// Envelope delivered to the conversation.
239    Sent(Envelope),
240    /// Envelope delivered from the conversation to a receiver.
241    Received(Envelope),
242    /// Participant crash handled according to the configured policy.
243    ParticipantCrashed {
244        /// Crashed participant PID.
245        participant: ParticipantPid,
246        /// Policy applied by the actor.
247        policy: CrashPolicy,
248    },
249    /// The conversation's supervision structure was violated: its exit watcher
250    /// died while the conversation was live (nothing legitimate terminates the
251    /// watcher externally). The conversation is failed and finalized in
252    /// response, never continued unobserved.
253    SupervisionFailed {
254        /// The dead watcher's PID.
255        watcher: ParticipantPid,
256        /// EXIT reason carried by the watcher's exit signal, when one was.
257        reason: Option<ExitReason>,
258    },
259}
260
261/// Queryable state snapshot owned by a conversation actor.
262#[derive(Clone, Debug, PartialEq, Eq)]
263pub struct ConversationState {
264    /// Current lifecycle phase.
265    pub current_phase: ConversationPhase,
266    /// Accumulated exchanged messages and lifecycle references.
267    pub context: Vec<ConversationContextEntry>,
268    /// Optional in-memory deadline.
269    pub deadline: Option<Instant>,
270    /// Last known participant liveness.
271    pub participants: Vec<ParticipantStatus>,
272    /// Durability mode marker retained for diagnostics and future resume work.
273    pub mode: ChannelMode,
274}
275
276impl ConversationState {
277    /// Creates initial state for a conversation config.
278    #[must_use]
279    pub fn from_config(config: &ConversationConfig, now: Instant) -> Self {
280        let deadline = config.timeout.map(|timeout| now + timeout);
281        let participants = config
282            .participants
283            .iter()
284            .copied()
285            .map(ParticipantStatus::alive)
286            .collect();
287
288        Self {
289            current_phase: ConversationPhase::Created,
290            context: Vec::new(),
291            deadline,
292            participants,
293            mode: config.mode,
294        }
295    }
296
297    /// Transitions Created to Active. Active is accepted as idempotent.
298    ///
299    /// # Errors
300    ///
301    /// Returns [`LiminalError::ConversationFailed`] when the state cannot become active.
302    pub fn activate(&mut self) -> Result<(), LiminalError> {
303        match self.current_phase {
304            ConversationPhase::Created => {
305                self.current_phase = ConversationPhase::Active;
306                Ok(())
307            }
308            ConversationPhase::Active => Ok(()),
309            phase => Err(invalid_transition(phase, ConversationPhase::Active)),
310        }
311    }
312
313    /// Transitions Active to Completing.
314    ///
315    /// # Errors
316    ///
317    /// Returns [`LiminalError::ConversationFailed`] when the state is not active.
318    pub fn begin_completing(&mut self) -> Result<(), LiminalError> {
319        match self.current_phase {
320            ConversationPhase::Active => {
321                self.current_phase = ConversationPhase::Completing;
322                Ok(())
323            }
324            ConversationPhase::Completing => Ok(()),
325            phase => Err(invalid_transition(phase, ConversationPhase::Completing)),
326        }
327    }
328
329    /// Transitions Completing to Closed.
330    ///
331    /// # Errors
332    ///
333    /// Returns [`LiminalError::ConversationFailed`] when completion has not begun.
334    pub fn close(&mut self) -> Result<(), LiminalError> {
335        match self.current_phase {
336            ConversationPhase::Completing => {
337                self.current_phase = ConversationPhase::Closed;
338                Ok(())
339            }
340            ConversationPhase::Closed => Ok(()),
341            phase => Err(invalid_transition(phase, ConversationPhase::Closed)),
342        }
343    }
344
345    /// Transitions any phase to Failed.
346    pub const fn fail(&mut self) {
347        self.current_phase = ConversationPhase::Failed;
348    }
349
350    /// Records an envelope sent into the conversation.
351    pub fn record_sent(&mut self, envelope: Envelope) {
352        self.context.push(ConversationContextEntry::Sent(envelope));
353    }
354
355    /// Records an envelope received from the conversation.
356    pub fn record_received(&mut self, envelope: Envelope) {
357        self.context
358            .push(ConversationContextEntry::Received(envelope));
359    }
360
361    /// Records participant crash handling and marks that participant dead,
362    /// stamping `exited_at` as the instant the EXIT signal was observed and
363    /// `reason` as the signal's exit reason (when one was carried).
364    pub fn record_participant_crash(
365        &mut self,
366        participant: ParticipantPid,
367        policy: CrashPolicy,
368        exited_at: Instant,
369        reason: Option<ExitReason>,
370    ) {
371        for status in &mut self.participants {
372            if status.participant == participant {
373                status.mark_dead_at(exited_at, reason);
374            }
375        }
376        self.context
377            .push(ConversationContextEntry::ParticipantCrashed {
378                participant,
379                policy,
380            });
381    }
382}
383
384/// Cloneable handle for interacting with a conversation actor.
385#[derive(Clone, Debug)]
386pub struct ConversationHandle {
387    backend: Arc<dyn ConversationHandleBackend>,
388}
389
390impl ConversationHandle {
391    pub(crate) fn new(backend: Arc<dyn ConversationHandleBackend>) -> Self {
392        Self { backend }
393    }
394
395    /// Sends a message envelope to the conversation actor.
396    ///
397    /// # Errors
398    ///
399    /// Returns a [`LiminalError`] when the actor cannot accept or process the message.
400    pub fn send(&self, message: impl Into<Envelope>) -> Result<(), LiminalError> {
401        self.backend.send(message.into())
402    }
403
404    /// Receives the next available envelope from the conversation actor.
405    ///
406    /// # Errors
407    ///
408    /// Returns a [`LiminalError`] when the actor is closed, failed, or unavailable.
409    pub fn receive(&self) -> Result<Envelope, LiminalError> {
410        self.backend.receive()
411    }
412
413    /// Closes the conversation normally.
414    ///
415    /// # Errors
416    ///
417    /// Returns a [`LiminalError`] when the actor cannot close normally.
418    pub fn close(&self) -> Result<(), LiminalError> {
419        self.backend.close()
420    }
421
422    /// Queries the actor state for diagnostics.
423    ///
424    /// # Errors
425    ///
426    /// Returns a [`LiminalError`] when the state cannot be queried.
427    pub fn query_state(&self) -> Result<ConversationState, LiminalError> {
428        self.backend.query_state()
429    }
430
431    /// Returns the current actor PID.
432    ///
433    /// # Errors
434    ///
435    /// Returns a [`LiminalError`] when the supervisor cannot inspect the actor.
436    pub fn actor_pid(&self) -> Result<ParticipantPid, LiminalError> {
437        self.backend.actor_pid()
438    }
439}
440
441pub(crate) trait ConversationHandleBackend: std::fmt::Debug + Send + Sync {
442    fn send(&self, message: Envelope) -> Result<(), LiminalError>;
443    fn receive(&self) -> Result<Envelope, LiminalError>;
444    fn close(&self) -> Result<(), LiminalError>;
445    fn query_state(&self) -> Result<ConversationState, LiminalError>;
446    fn actor_pid(&self) -> Result<ParticipantPid, LiminalError>;
447}
448
449fn invalid_transition(from: ConversationPhase, to: ConversationPhase) -> LiminalError {
450    LiminalError::ConversationFailed {
451        message: format!("invalid conversation phase transition from {from:?} to {to:?}"),
452    }
453}
454
455#[cfg(test)]
456mod tests {
457    use super::{Conversation, ConversationHandle};
458
459    #[test]
460    fn starting_conversation_creates_named_span_with_fresh_trace_context() {
461        let first = Conversation::start("conversation-1");
462        let second = Conversation::start("conversation-2");
463
464        assert_eq!(first.name(), "conversation-1");
465        assert_eq!(first.parent_trace_context(), None);
466        assert_ne!(first.trace_context().trace_id(), 0);
467        assert_ne!(first.trace_context().span_id(), 0);
468        assert_ne!(
469            first.trace_context().trace_id(),
470            second.trace_context().trace_id()
471        );
472    }
473
474    #[test]
475    fn messages_inherit_conversation_trace_context_automatically() {
476        let conversation = Conversation::start("conversation");
477        let message = conversation.message("payload");
478
479        assert_eq!(message.payload(), &"payload");
480        assert_eq!(message.trace_context(), conversation.trace_context());
481    }
482
483    #[test]
484    fn child_conversation_references_parent_trace_context() {
485        let parent = Conversation::start("parent");
486        let child = parent.spawn_child("child");
487
488        assert_eq!(child.name(), "child");
489        assert_eq!(child.parent_trace_context(), Some(parent.trace_context()));
490        assert_eq!(
491            child.trace_context().trace_id(),
492            parent.trace_context().trace_id()
493        );
494        assert_ne!(
495            child.trace_context().span_id(),
496            parent.trace_context().span_id()
497        );
498    }
499
500    #[test]
501    fn message_mapping_preserves_trace_context() {
502        let conversation = Conversation::start("conversation");
503        let context = conversation.trace_context();
504        let mapped = conversation.message(1_u8).map(u16::from);
505
506        assert_eq!(mapped.payload(), &1_u16);
507        assert_eq!(mapped.trace_context(), context);
508    }
509
510    #[test]
511    fn conversation_handle_is_clone_send_sync() {
512        fn assert_clone_send_sync<T: Clone + Send + Sync>() {}
513
514        assert_clone_send_sync::<ConversationHandle>();
515    }
516}
liminal/conversation/types.rs

liminal/conversation/
types.rs
