liminal-rs 0.16.5

//! Host-side reply waiting, answered by the reply or by the actor's death (LIM-009).
//!
//! A command's reply [`mpsc::SyncSender`] is moved into the actor's command
//! queue, which OUTLIVES the beamr process. The supervisor's exit watcher is
//! told of every channel actor's exit and retires its queue, which drops the
//! abandoned senders (`ActorRuntime::retire`). So a waiter needs nothing but one
//! bounded receive: the reply arrives, or the death disconnects it at once, and
//! [`COMMAND_TIMEOUT`] bounds only a live actor that never answers.

use std::sync::mpsc;
use std::time::Duration;

use crate::channel::schema::{SchemaId, SchemaValidationError};
use crate::error::LiminalError;

/// Default bound on how long a handle blocks for a command reply.
pub(super) const COMMAND_TIMEOUT: Duration = Duration::from_secs(5);

/// Why a reply wait ended without the actor replying.
enum WaitFailure {
    /// The actor died before replying: its retirement dropped the reply sender.
    Dead,
    /// The full command timeout elapsed with no reply and no death.
    TimedOut,
}

/// Block for a command reply. Surfaces a [`LiminalError`] for the standard
/// command reply path.
pub(super) fn wait_live<T>(
    response: &mpsc::Receiver<Result<T, LiminalError>>,
    pid: u64,
) -> Result<Result<T, LiminalError>, LiminalError> {
    receive(response).map_err(|failure| match failure {
        WaitFailure::Dead => LiminalError::DeliveryFailed {
            message: format!("channel actor pid {pid} died before replying"),
        },
        WaitFailure::TimedOut => LiminalError::DeliveryFailed {
            message: "channel command timed out".to_owned(),
        },
    })
}

/// Schema-evolution variant of [`wait_live`]: surfaces a
/// [`SchemaValidationError`] for the schema-typed reply path.
pub(super) fn wait_schema_live(
    response: &mpsc::Receiver<Result<SchemaId, SchemaValidationError>>,
    pid: u64,
) -> Result<Result<SchemaId, SchemaValidationError>, SchemaValidationError> {
    receive(response).map_err(|failure| match failure {
        WaitFailure::Dead => SchemaValidationError::InvalidSchema {
            message: format!("channel actor pid {pid} died before evolving the schema"),
        },
        WaitFailure::TimedOut => SchemaValidationError::InvalidSchema {
            message: "channel schema-evolution reply unavailable".to_owned(),
        },
    })
}

/// One bounded receive: the reply, or the sender's drop by the actor's retirement.
fn receive<R>(response: &mpsc::Receiver<R>) -> Result<R, WaitFailure> {
    response
        .recv_timeout(COMMAND_TIMEOUT)
        .map_err(|error| match error {
            mpsc::RecvTimeoutError::Disconnected => WaitFailure::Dead,
            mpsc::RecvTimeoutError::Timeout => WaitFailure::TimedOut,
        })
}