dbos 0.6.0

//! Durable sleep: a wait that survives the process waiting it out.

use std::sync::Arc;
use std::time::Duration;

use crate::checkpoint::{PendingStep, StepPlacement, StepWriter};
use crate::error::{DurableError, Error};
use crate::sysdb::types::{StepCaller, duration_to_ms, step_names};

/// Waits for `duration`, resuming at the **original** wake time after a crash.
///
/// The difference from [`tokio::time::sleep`] is what happens when the process dies. A plain sleep
/// restarts from the beginning on recovery, so a workflow that sleeps an hour and crashes after
/// fifty minutes sleeps another hour. This records the wake time as a checkpoint before waiting, so
/// the same workflow has ten minutes left — which is the only version of "sleep for an hour" that
/// composes with durable execution.
///
/// ```no_run
/// # async fn f() -> dbos::Result<()> {
/// use std::time::Duration;
/// dbos::sleep(Duration::from_secs(3600)).await?;
/// # Ok(()) }
/// ```
///
/// **The step id is taken here, at the call, not at the first poll** — see [`PendingStep`]. A sleep
/// built beside a step and driven with it by `tokio::join!` therefore takes the same slot on every
/// execution, however the two futures interleave, and a sleep built and dropped has spent that slot
/// all the same.
///
/// **Outside a workflow this is a plain sleep**, the way a step outside a workflow is a plain call:
/// there is no step sequence to checkpoint against, and a function built from steps and sleeps
/// stays ordinarily callable and ordinarily testable. Inside a *step* the same applies, since a
/// step is a leaf.
///
/// The checkpoint's `completed_at` is stamped at the wake time rather than at the moment the row is
/// written, so an hour's sleep reads as an hour on a timeline instead of as an instant.
///
/// A zero or negative duration is not an error: it records the checkpoint and returns, so a
/// computed delay that has already elapsed behaves the same on the first run and on a replay.
///
/// **A duration too long to record is [`Error::InvalidArgument`]**: one that does not fit in
/// milliseconds as a 64-bit integer. Refused at the call, before the step id is taken, so a refused
/// sleep moves no counter — and refused outside a workflow too, so the same call does not succeed
/// or fail depending on where it is written.
///
/// A duration that fits but carries the wake time past the largest instant those milliseconds can
/// hold is refused by the checkpoint instead, after the step id is taken, as the same
/// [`Error::InvalidArgument`]. That rule reads the clock, so it is applied only where the record
/// has been looked for first, and the refusal is recorded like a wake time: a replay gets back
/// what the first run recorded, either one, rather than deciding again against a later clock.
pub fn sleep<E>(duration: Duration) -> PendingStep<'static, (), E>
where
    E: DurableError + 'static,
{
    // The whole of what happens at the call: read where we are, and take the id if this is a place
    // that checkpoints. A sleep is always served by the workflow it is written in, so the
    // placement has no second connection to disagree with and cannot fail — it answers `Outside`
    // for itself, which is the one case that has no executor to carry either.
    //
    // The executor comes off the placement rather than from a second read of the ambient context,
    // so the two cannot disagree about whether there is one: that is the invariant the run below
    // leans on when it asks for an executor and an id together.
    //
    // The duration is checked first, so a refusal claims no id — and only against a rule that
    // reads no clock, so the first run and a replay decide it the same way.
    let built = match duration_to_ms("duration", duration) {
        Err(error) => Err(Error::refused_argument("sleep", error)),
        Ok(_) => {
            let placement = StepPlacement::here();
            Ok((placement.executor().map(Arc::clone), placement))
        }
    };
    PendingStep::placed(
        step_names::SLEEP,
        built,
        move |executor, placement| async move {
            // Both halves or neither: an executor with no id behind it is the in-step and
            // outside-a-workflow cases, where the placement recorded nothing and there is
            // nothing to record now.
            let Some((
                executor,
                StepWriter {
                    workflow_id,
                    step_id,
                    owner_xid,
                },
            )) = executor.zip(placement.step_writer())
            else {
                tracing::debug!(
                    duration_ms = duration.as_millis(),
                    "the sleep is not checkpointed: it is outside a workflow, or inside a step"
                );
                crate::rt::sleep(duration).await;
                return Ok(());
            };

            let wake = executor
                .sysdb()
                .record_sleep(
                    StepCaller {
                        workflow_id,
                        step_id,
                        codec: executor.serializer(),
                        owner_xid,
                    },
                    duration,
                )
                .await
                .map_err(|error| Error::refused_argument("sleep", error))?;

            // From the recorded wake time, not from now: on a replay `record_sleep` gives back the
            // instant the first run chose, and the remaining wait is whatever is left of it. A
            // replay that woke long ago waits not at all. Measured on the monotonic clock from
            // the database's reading, never against this host's wall clock, which need not agree
            // with the clock the wake time was set on.
            let until = wake.arrives();
            tracing::debug!(
                step_id,
                remaining_ms = until
                    .saturating_duration_since(crate::rt::now())
                    .as_millis(),
                "sleeping until the recorded wake time"
            );
            crate::rt::sleep_until(until).await;
            Ok(())
        },
    )
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Outside a workflow too, so the same call does not succeed in one place and fail in another.
    #[tokio::test]
    async fn a_sleep_that_cannot_be_recorded_is_refused_outside_a_workflow() {
        let refused = sleep::<crate::error::EngineOnly>(Duration::MAX).await;
        match refused {
            Err(Error::InvalidArgument { operation, detail }) => {
                assert_eq!(operation, "sleep");
                assert_eq!(
                    detail,
                    "`duration` must fit in milliseconds as a 64-bit integer"
                );
            }
            other => panic!("expected the sleep to be refused, got {other:?}"),
        }
    }
}