dbos 0.5.0

//! Steps: the checkpoints that make a workflow resumable.

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

use serde::Serialize;
use serde::de::DeserializeOwned;
use tracing::Instrument;

use tokio_util::sync::CancellationToken;

use crate::checkpoint::{PendingStep, StepDurability, StepPlacement, revive};
use crate::context::{Ctx, StepStatus};
use crate::error::{DurableError, EngineOnly, Error, Result};
use crate::serialization::{decode, encode};
use crate::sysdb::types::{AwaitedOutcome, Outcome, StepTiming, Timestamp};

/// A step's retry predicate: given a failure, whether to try again.
///
/// Named as a type because it appears in [`StepOptions::should_retry`]'s field, where the bare
/// `Arc<dyn Fn(..)>` is more punctuation than signature. Callers rarely write it —
/// [`StepOptions::should_retry`](StepOptions::should_retry) takes the closure and wraps it.
pub type ShouldRetry<E> = Arc<dyn Fn(&Error<E>) -> bool + Send + Sync>;

/// What a caller may say about a step, beyond its name and body.
///
/// A struct rather than a builder, matching [`StartOptions`](crate::StartOptions) and how `sysdb`
/// spells optional arguments. The common case pays nothing because [`step`] keeps its no-option
/// form. Deliberately **not** `#[non_exhaustive]`, so `..Default::default()` keeps working as
/// fields are added.
///
/// **The defaults are the three-way majority, not Go's.** Python, TypeScript and Java all use a
/// one-second base with a 2.0 rate; Go uses 100ms with a five-second cap, which is a different
/// feature — that one is for a flaky network call, this one is for an upstream that is down.
///
/// **The predicate is synchronous**, unlike Python's and TypeScript's, which also accept an
/// `async` one. Go's `WithStepRetryPredicate` and Java's `StepShouldRetry` are both synchronous, so
/// this is 2-of-4 rather than a deviation from consensus — and in Rust the two are not equally
/// priced. An async predicate cannot be stored in a struct field without boxing its future, so
/// every caller would write `Arc::new(|e| Box::pin(async move { .. }))` to express what is almost
/// always a pure match on an error. A check that genuinely has to await belongs in the body, which
/// can return an error the predicate then declines.
pub struct StepOptions<E = EngineOnly> {
    /// How many times the body may run before the step is failed. Defaults to **1**: no retrying.
    ///
    /// All four references agree that a plain step does not retry, and spell it three different
    /// ways — Python and TypeScript with a `retries_allowed` boolean beside a count, Go with
    /// `maxRetries == 0`, Java by flooring `maxAttempts` at 1. The count alone is taken here
    /// because a boolean beside it can express `retries_allowed: false, max_attempts: 5`, which
    /// means nothing, and an option that cannot state a contradiction is worth more than one that
    /// matches a reference's field list.
    ///
    /// Zero is treated as one: a step whose body never runs is not a step.
    pub max_attempts: u32,
    /// How long to wait after the first failed attempt.
    pub interval: Duration,
    /// What the interval is multiplied by after each failure.
    pub backoff_rate: f64,
    /// The longest the interval may grow to, however many failures precede it.
    pub max_interval: Duration,
    /// The longest **one attempt** may run. `None` lets it run as long as it likes.
    ///
    /// On expiry the attempt's future is dropped — which is how Rust stops work, and stops it at
    /// the next suspension point while running destructors on the way out — and the step fails
    /// with [`Error::StepTimeout`]. That is an ordinary retryable failure: it is offered to
    /// [`should_retry`](Self::should_retry) and counts against
    /// [`max_attempts`](Self::max_attempts), as TypeScript's `timeoutMS` also specifies.
    ///
    /// **Each attempt gets its own timeout, and backoff is not charged against it.** Three
    /// attempts at five seconds may spend fifteen seconds in the body. Python states the same
    /// layering where it puts the supervisor inside the retry loop.
    ///
    /// Before the future is dropped, the attempt's cancellation token fires, which
    /// [`cancellation_token`](crate::cancellation_token) hands to the body — the way to stop work
    /// that dropping the future cannot reach, such as a blocking thread. Ordinary `async` bodies
    /// need none of it: they stop at their next suspension point and run their destructors,
    /// returning connections and releasing guards without the body saying so.
    ///
    /// **Unlike Python, this is not restricted to some kinds of step.** py #826 rejects a timeout
    /// on a sync step because *"Python has no preemption mechanism for sync steps"*; every step
    /// here is a future, so there is no second case to exclude and no error to raise.
    pub timeout: Option<Duration>,
    /// Whether to stop this step as soon as the workflow is seen to be `CANCELLED` elsewhere.
    ///
    /// Off by default, and worth turning on for a long step whose work is wasted once the workflow
    /// is cancelled — a large transfer, a slow report. A cancellation raised in **this** process
    /// does not need it, since that path stops the workflow directly; what this observes is a
    /// cancellation raised somewhere else, by Conductor, a client, or another SDK sharing the
    /// system database.
    ///
    /// The cost is a periodic status read per running step, at
    /// [`Config::outcome_poll_interval`](field@crate::Config::outcome_poll_interval), under the same
    /// polling-concurrency cap as every other database-backed wait. Python's `preemptible` does
    /// the same and hardcodes its interval.
    ///
    /// **A preempted step records nothing** and runs again on resume, because a cancellation is a
    /// control signal rather than the step's result — the step did not fail, it was interrupted.
    pub preemptible: bool,
    /// Decides whether a failure is worth retrying. `None` retries every failure.
    ///
    /// Returning `false` ends the step immediately with that error, even with attempts left — so a
    /// 4xx from an API is not hammered three times while a 5xx is. Named for the majority: Python's
    /// `should_retry`, TypeScript's `shouldRetry` and Java's `StepShouldRetry` agree, and only Go
    /// spells it `WithStepRetryPredicate`.
    ///
    /// **It is evaluated before the backoff sleep**, so declining an error costs no wait. That is
    /// Go's documented behaviour and the only sensible order: waiting to find out whether to wait
    /// helps nobody.
    ///
    /// **A control signal never reaches it.** Cancellation, shutdown and system-database failures
    /// end the step before the policy is consulted, so a predicate cannot elect to retry against a
    /// database that is down, and does not need an arm for a case it will never see. Use
    /// [`should_retry`](Self::should_retry) rather than writing the `Arc` by hand.
    pub should_retry: Option<ShouldRetry<E>>,
}

impl<E> Default for StepOptions<E> {
    fn default() -> Self {
        Self {
            max_attempts: 1,
            interval: Duration::from_secs(1),
            backoff_rate: 2.0,
            max_interval: Duration::from_secs(3600),
            timeout: None,
            preemptible: false,
            should_retry: None,
        }
    }
}

/// Hand-written because a closure is not [`Debug`], and deriving would demand `E: Debug` besides.
impl<E> std::fmt::Debug for StepOptions<E> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("StepOptions")
            .field("max_attempts", &self.max_attempts)
            .field("interval", &self.interval)
            .field("backoff_rate", &self.backoff_rate)
            .field("max_interval", &self.max_interval)
            .field("timeout", &self.timeout)
            .field("preemptible", &self.preemptible)
            .field("should_retry", &self.should_retry.is_some())
            .finish()
    }
}

/// Hand-written because deriving would demand `E: Clone`, which no caller owes.
impl<E> Clone for StepOptions<E> {
    fn clone(&self) -> Self {
        Self {
            max_attempts: self.max_attempts,
            interval: self.interval,
            backoff_rate: self.backoff_rate,
            max_interval: self.max_interval,
            timeout: self.timeout,
            preemptible: self.preemptible,
            should_retry: self.should_retry.clone(),
        }
    }
}

impl<E> StepOptions<E> {
    /// Sets [`should_retry`](Self::should_retry), wrapping the closure.
    ///
    /// A method beside the struct literal rather than instead of it: the other fields are
    /// plain values and read better set directly, while this one would otherwise make every call
    /// site spell `Some(Arc::new(..))` around a one-line match.
    ///
    /// ```no_run
    /// # #[derive(Debug, thiserror::Error, serde::Serialize, serde::Deserialize)]
    /// # #[error("boom")] struct ApiError { status: u16 }
    /// use dbos::StepOptions;
    /// let options = StepOptions {
    ///     max_attempts: 3,
    ///     ..Default::default()
    /// }
    /// .should_retry(|error| match error {
    ///     dbos::Error::Application(ApiError { status }) => *status >= 500,
    ///     _ => true,
    /// });
    /// ```
    pub fn should_retry(
        mut self,
        predicate: impl Fn(&Error<E>) -> bool + Send + Sync + 'static,
    ) -> Self {
        self.should_retry = Some(Arc::new(predicate));
        self
    }

    /// The wait before the attempt following `failures` failures.
    ///
    /// `interval * rate^failures`, capped. Saturating rather than panicking on a nonsensical rate:
    /// a negative or `NaN` `backoff_rate` yields a non-finite or negative product, and a large one
    /// yields a product past what a `Duration` can hold — all three go to the cap rather than to a
    /// constructor that would panic. A retry that waits too long is a misconfiguration; a retry
    /// that aborts the process is a bug.
    ///
    /// **The cap is applied to the `f64`, not to the `Duration` built from it.** Clamping
    /// afterwards reads the same and is not: `Duration::from_secs_f64` panics on any value that
    /// overflows a `Duration`, so `interval: 1s, backoff_rate: 1e10` reaches the panic on its
    /// second failure with the clamp sitting unreachable behind it.
    fn backoff(&self, failures: u32) -> Duration {
        let grown = self.interval.as_secs_f64() * self.backoff_rate.powi(failures as i32);
        if !grown.is_finite() || grown < 0.0 || grown >= self.max_interval.as_secs_f64() {
            return self.max_interval;
        }
        Duration::from_secs_f64(grown)
    }
}

/// Runs `body` once per workflow, recording what it returned.
///
/// On a first run the body executes and its result is written to the database. On a replay the
/// recorded result is returned **without entering the body at all** — which is the whole of durable
/// execution, and why a step may do things that must not happen twice.
///
/// ```no_run
/// # async fn f() -> dbos::Result<()> {
/// let charge = dbos::step("charge_card", || async { dbos::Result::Ok(42) }).await?;
/// # Ok(()) }
/// ```
///
/// The error type is a type parameter so that a recorded failure can be decoded back into it. It
/// is the *workflow's* error type rather than one of the step's own, which is what lets an
/// infallible body write a bare `Ok(..)`: there is nothing to infer, because the surrounding `?`
/// already fixed it. A step failing with some other error type converts at the boundary, with
/// `.map_err(MyError::from)?`.
///
/// Outside a workflow the body simply runs, undurably. That makes a function built from steps
/// ordinarily callable and ordinarily testable, and it is what Python does. Inside another step the
/// same applies: a step is a leaf, so a nested one is a plain call rather than a second checkpoint.
///
/// **Steps may run concurrently, and the id is what makes that sound.** This call takes the id
/// from the workflow's counter *here*, in the caller's own sequential order, and hands back a
/// [`PendingStep`] that has not run — so a set of steps built and then driven together gets the same
/// slots on a replay however their bodies interleave. `tokio::join!` over steps is therefore
/// ordinary code: it builds every branch before polling any, which is exactly the order the ids
/// were taken in. An id allocated at the first poll would instead depend on which future reached
/// the counter first, which is not something a replay reproduces.
///
/// **That is a promise about every durable call in the crate.** `sleep`, the events, the
/// messages, the waits and every checkpointed management call on [`DBOS`](crate::DBOS) take their
/// ids at the call, and so do a child's [`start`](crate::WorkflowRef::start), the await of its
/// handle ([`WorkflowHandle::result`](crate::WorkflowHandle::result)) and the
/// [`run`](crate::WorkflowRef::run) that is the two of them in sequence. Any of them may be built
/// first and driven together with steps and with each other; `join!` over a mixture of them is
/// ordinary code.
///
/// **`join!` needs nothing of a race; a race needs its winner recorded.** An all-wait decides
/// nothing, so a replay has nothing to get differently and `join!` over durable calls is ordinary
/// code. A race decides something — which branch won — and that decision has to survive to the
/// replay like any other, or the replay is free to decide it the other way and continue from a
/// branch this execution abandoned. So what a workflow body has no use for is the *unrecorded*
/// race: a bare `tokio::select!` over durable calls, and `tokio::time::timeout`, which is that
/// same race against a clock no replay reproduces.
///
/// Recorded races are the supported ones. [`select_step!`](crate::select_step) is the general one:
/// it checkpoints *which branch* won, and a replay polls that branch alone.
/// [`select_workflow!`](macro@crate::select_workflow) is the specialised one, for a race whose
/// branches are all workflows: it checkpoints which handle won and settles the whole set with one
/// wait. A call that carries its own deadline is the third — there the bound is part of what gets
/// recorded. And a race over futures this crate knows nothing about can still go *inside a step*,
/// whose checkpoint stands for however its body reached the answer. [`PendingStep`] says all of
/// this for every durable call, not only steps; ids taken at the call are what let a race be
/// recorded at all, since a losing branch spends the same id on the replay whether or not it is
/// ever polled.
///
/// **Whether a step is nested is decided per call stack, not per workflow.** The context a step
/// body runs under is rebound for that body alone, so a step built in the workflow proper while a
/// sibling's body is in flight still takes an id and checkpoints.
///
/// **A step built and dropped has still spent its id**, which is why [`PendingStep`] is `#[must_use]`.
/// It is deterministic — the same construction sequence burns the same ids on a replay — but it is
/// not a no-op.
///
/// **A step is polled where it was built.** The id is a claim on one position in one workflow, so a
/// step built outside a workflow and awaited inside one, carried into a second workflow, or carried
/// across a step-body boundary in either direction is refused as [`Error::StepBuiltElsewhere`]
/// rather than run under an id nothing there can honour. The everyday way to trip it is building
/// the step before the workflow runs — in the handler that then starts it, say — and awaiting it
/// inside.
///
/// The name is explicit and it matters: it is checked on replay, so a step whose name changed is
/// reported rather than silently matched against the recorded result of whatever used to be there.
///
/// **The body is `FnMut` rather than `FnOnce` because a step may be attempted more than once.**
/// Retries call it again, and a bound that permits exactly one call cannot express that. The cost
/// to a caller is nothing in the ordinary case — a closure written inline at the call site is
/// `FnMut` unless it moves a captured value out — and a body that genuinely consumes what it
/// captured fails to compile here rather than at its second attempt, which is where the mistake
/// should be reported.
pub fn step<'a, T, E, F, Fut>(name: &str, body: F) -> PendingStep<'a, T, E>
where
    T: Serialize + DeserializeOwned + Send + 'a,
    E: DurableError + Send,
    F: FnMut() -> Fut + Send + 'a,
    Fut: Future<Output = Result<T, E>> + Send + 'a,
{
    step_with(name, StepOptions::default(), body)
}

/// Runs `body` as a step, retrying it as `options` allows.
///
/// [`step`] with a retry policy — one mechanism, and [`step`] is this called with the defaults.
///
/// ```no_run
/// # async fn f() -> dbos::Result<()> {
/// use dbos::StepOptions;
/// let charge = dbos::step_with(
///     "charge_card",
///     StepOptions { max_attempts: 3, ..Default::default() },
///     || async { dbos::Result::Ok(42) },
/// )
/// .await?;
/// # Ok(()) }
/// ```
///
/// **Every attempt shares one checkpoint.** The recorded result is looked for once, before the
/// first attempt, and written once, after the last; a replay of a step that took three attempts
/// sees one outcome, and the intermediate failures live only in logs and spans. That is why
/// `operation_outputs` has no attempt column in any of the five implementations.
///
/// **A control signal is never retried.** Cancellation, shutdown, and any system-database failure
/// end the step immediately with nothing recorded, so the workflow stays `PENDING` and is
/// recovered — a database blip is not evidence that the body is wrong, and retrying against a
/// database that is down would burn the whole policy before the first useful attempt.
///
/// The recorded `started_at` covers the **whole sequence**, from before the recorded-result check
/// to after the final attempt, rather than the last attempt alone. Python takes its
/// `step_start_time` in the same place, and so does Go (#442).
pub fn step_with<'a, T, E, F, Fut>(
    name: &str,
    options: StepOptions<E>,
    body: F,
) -> PendingStep<'a, T, E>
where
    T: Serialize + DeserializeOwned + Send + 'a,
    E: DurableError + Send,
    F: FnMut() -> Fut + Send + 'a,
    Fut: Future<Output = Result<T, E>> + Send + 'a,
{
    // **The whole of what happens at the call**: read where we are, and take the id if this is a
    // place that checkpoints. Everything below is deferred into the run, which does not begin
    // until something polls it.
    let placement = StepPlacement::here();
    let name: Arc<str> = Arc::from(name);
    // The placement is held by both halves: the run needs the answer it gives, and the
    // `PendingStep` needs to ask it again on every poll. Cloning it is a couple of `Arc` bumps.
    PendingStep::new(
        Arc::clone(&name),
        placement.clone(),
        run(placement, name, options, body),
    )
}

/// The step itself, once something polls it.
///
/// **The first thing it does is check that it is where it was built**, because the id it carries is
/// a claim on one position in one workflow and nothing else can honour it. That check is
/// [`StepPlacement::check_here`] rather than anything written here: the rule belongs to the id, so
/// every call that takes one asks the same question of the same code.
async fn run<T, E, F, Fut>(
    placement: StepPlacement,
    name: Arc<str>,
    options: StepOptions<E>,
    mut body: F,
) -> Result<T, E>
where
    T: Serialize + DeserializeOwned,
    E: DurableError,
    F: FnMut() -> Fut,
    Fut: Future<Output = Result<T, E>>,
{
    let name = &*name;
    // Borrowed, not cloned: the answer is about where this stands, and `StepDurability` borrows
    // the placement rather than the ambient context, so nothing here needs to own one.
    let (ctx, step_id) = match Ctx::with_current(|here| placement.check_here(name, here))? {
        StepDurability::Recorded { ctx, step_id } => (ctx, step_id),
        // Outside a workflow, inside another step's body, or on a client's connection. The step
        // claimed no id in any of them, so there is nothing to check against and nothing to
        // record; `built` names which, rather than this deciding again.
        StepDurability::Plain => {
            tracing::debug!(
                step_name = name,
                built = %placement.whereabouts(),
                "the step body runs plainly"
            );
            return body().await;
        }
    };

    let executor = ctx.executor();
    let workflow_id = ctx.workflow_id();

    // Before the check, not after it: the recorded duration covers the whole step, including the
    // round trip that asks whether it has already run. Python takes `step_start_time` here, and
    // so does Go (#442).
    let started_at = Timestamp::now();

    let recorded = executor
        .sysdb()
        .check_step(workflow_id, step_id, name)
        .await
        .map_err(Error::SystemDatabase)?;
    if let Some(recorded) = recorded {
        tracing::debug!(
            step_id,
            step_name = name,
            "the step replays from its checkpoint; the body does not run"
        );
        return match recorded.error {
            Some(error) => Err(revive(&error, name)),
            None => decode(recorded.output.as_deref(), "step result"),
        };
    }

    let attempts = options.max_attempts.max(1);
    // Only ever pushed to when retrying, so a step at the default of one attempt allocates nothing.
    let mut failures: Vec<Error<E>> = Vec::new();
    let outcome = loop {
        let attempt = failures.len() as u32 + 1;
        // Rebuilt per attempt, because `current_attempt` is the field that moves: the id and the
        // cap are the step's, the attempt number is this try's.
        let status = StepStatus {
            step_id,
            current_attempt: attempt,
            max_attempts: attempts,
        };
        // The span nests inside the workflow's, so anything the body logs carries both ids.
        let span = tracing::info_span!("step", step_id, step_name = name, attempt);
        // **The closure is called inside the step's scope, not beside it.** An `async` block
        // does nothing until it is polled, and `in_step_scope` is what polls this one, so a body
        // that *builds* a durable call as its first act — `|| step("inner", ..)`, or a helper
        // that hands one back — makes that call with this attempt's marker in scope and is a
        // plain nested call by the leaf rule. Passing `body()` here instead would run the closure
        // on the workflow's own call stack, where it takes a step id from the counter, and the
        // call would then be refused by its own run for being polled inside a step body. Every
        // attempt would spend another id on the way.
        match supervise(ctx, name, status, &options, async { body().await }, span).await {
            Ok(value) => break Ok(value),
            // Not the step's result and not retryable: a cancelled workflow, a shutdown, or a
            // database that is down says nothing about whether the body would succeed. Returning
            // here leaves the row untouched, so the workflow stays `PENDING` and is recovered.
            Err(error) if error.control().is_some() => {
                tracing::debug!(
                    step_id,
                    step_name = name,
                    attempt,
                    "a control signal ended the step; it is not retried and nothing is checkpointed"
                );
                return Err(error);
            }
            Err(error) if attempt >= attempts => {
                if failures.is_empty() {
                    // No retry policy to report on, so the error is the step's outcome as it
                    // stands. Wrapping here would put every ordinary failure inside a collection
                    // of one.
                    break Err(error);
                }
                failures.push(error);
                break Err(Error::MaxStepRetriesExceeded {
                    step: name.to_owned(),
                    attempts,
                    errors: std::mem::take(&mut failures),
                });
            }
            // Declined by the policy: this failure is the step's outcome, whatever attempts
            // remain. Checked before the backoff so refusing to retry costs no wait, which is Go's
            // documented order for the same hook.
            Err(error)
                if options
                    .should_retry
                    .as_ref()
                    .is_some_and(|should_retry| !should_retry(&error)) =>
            {
                tracing::debug!(
                    step_id,
                    step_name = name,
                    attempt,
                    error = %error,
                    "the retry predicate declined this failure; the step is not retried"
                );
                if failures.is_empty() {
                    break Err(error);
                }
                failures.push(error);
                break Err(Error::MaxStepRetriesExceeded {
                    step: name.to_owned(),
                    attempts: attempt,
                    errors: std::mem::take(&mut failures),
                });
            }
            Err(error) => {
                let backoff = options.backoff(attempt - 1);
                tracing::warn!(
                    step_id,
                    step_name = name,
                    attempt,
                    attempts,
                    backoff_secs = backoff.as_secs_f64(),
                    error = %error,
                    "the step failed and will be retried"
                );
                failures.push(error);
                tokio::time::sleep(backoff).await;
            }
        }
    };

    // Built once and held, not rebuilt per attempt: `record_step` compares the stored completion
    // time against this one to tell its own retried write from another execution's, and a fresh
    // timestamp on a retry would read as somebody else.
    let timing = Some(StepTiming {
        started_at,
        completed_at: Timestamp::now(),
    });
    let serialization = Some(executor.serializer().name());

    match &outcome {
        Ok(value) => {
            let output = encode(value, "step result")?;
            executor
                .sysdb()
                .record_step(
                    workflow_id,
                    step_id,
                    name,
                    Outcome::Output(Some(&output)),
                    serialization,
                    timing,
                )
                .await
                .map_err(Error::SystemDatabase)?;
            tracing::debug!(
                step_id,
                step_name = name,
                "the step ran; its output is recorded"
            );
        }
        Err(error) => {
            // A control error cannot arrive here: the loop above returns on one without leaving
            // itself, precisely so that a cancelled workflow's step is not recorded as *failed* and
            // replayed as permanently so. What reaches this arm is the step's own outcome.
            debug_assert!(error.control().is_none());
            // The error itself, encoded, not a description of it: a replay gives back the error
            // that failed exactly as it gives back the value that succeeded.
            let encoded = encode(error, "step error")?;
            executor
                .sysdb()
                .record_step(
                    workflow_id,
                    step_id,
                    name,
                    Outcome::Error(&encoded),
                    serialization,
                    timing,
                )
                .await
                .map_err(Error::SystemDatabase)?;
            tracing::debug!(
                step_id,
                step_name = name,
                "the step failed; its error is recorded"
            );
        }
    }
    outcome
}

/// Runs one attempt under the watchdogs its options ask for.
///
/// One place decides whether the attempt completed or blew its deadline, which is the shape of
/// Python's `_supervise_step` (py #826), where the preemption poller and the step timeout share
/// one loop. Preemption joins here rather than beside it.
///
/// **Without a timeout this is the body and nothing else** — no timer, no token, no `select!` —
/// so a step that does not ask for one pays nothing for the option existing.
///
/// With one, the attempt races a timer. On expiry the token fires *first* and the body's future is
/// then dropped, and the order matters: dropping stops the future at its next suspension point, so
/// anything that would only learn from the token has to be told before it goes. Dropping is also
/// the half TypeScript cannot do — it abandons a timed-out attempt and discards whatever the
/// abandoned promise eventually settles to, while a dropped Rust future stops *and* runs its
/// destructors, returning connections and releasing guards without the body saying so.
async fn supervise<T, E, Fut>(
    ctx: &Ctx,
    name: &str,
    status: StepStatus,
    options: &StepOptions<E>,
    body: Fut,
    span: tracing::Span,
) -> Result<T, E>
where
    E: DurableError,
    Fut: Future<Output = Result<T, E>>,
{
    // **Every attempt gets a token, and it fires wherever the attempt is abandoned.** The guard
    // cancels it if this future is dropped — a caller dropping the step, a combinator dropping it
    // as a losing branch, the `select!` below losing the attempt to a watchdog — so work the
    // runtime cannot stop by dropping, a `spawn_blocking` thread watching `cancellation_token()`,
    // learns that its step is over in every one of those cases and not only the two this function
    // races itself.
    //
    // **An attempt that returns disarms it**, which is what keeps the token a signal that the
    // step was abandoned rather than a step-is-over broadcast: the body reached an outcome and
    // had its chance to clean up, and work it deliberately left running is not this function's to
    // stop. The two watchdog arms still cancel explicitly, before the losing future is dropped at
    // the end of the `select!`, because a body that could only learn from the token should learn
    // before its destructors run.
    let token = CancellationToken::new();
    let cancel_on_drop = token.clone().drop_guard();
    let attempt = ctx
        .in_step_scope(token.clone(), status, body)
        .instrument(span);

    let outcome = if options.timeout.is_none() && !options.preemptible {
        attempt.await
    } else {
        supervised(ctx, name, options, attempt, token).await
    };
    cancel_on_drop.disarm();
    outcome
}

/// The watchdog half of [`supervise`], for an attempt that has a timeout or is preemptible.
///
/// Split out so the common case — neither watchdog — is an `await` on the attempt and nothing
/// else, and so the token's disarm has one home rather than one per return path.
async fn supervised<T, E>(
    ctx: &Ctx,
    name: &str,
    options: &StepOptions<E>,
    attempt: impl Future<Output = Result<T, E>>,
    token: CancellationToken,
) -> Result<T, E>
where
    E: DurableError,
{
    // A deadline that never arrives, so the arm can be unconditional rather than duplicating the
    // whole `select!` for each combination of watchdogs.
    let deadline = async {
        match options.timeout {
            Some(timeout) => tokio::time::sleep(timeout).await,
            None => std::future::pending().await,
        }
    };
    let cancelled = async {
        match options.preemptible {
            true => observe_cancellation(ctx).await,
            false => std::future::pending().await,
        }
    };

    // `attempt` is passed by value, so losing the race is what drops it — and dropping it is what
    // stops the body. Taking it by `&mut` under `tokio::pin!` would leave the future alive for the
    // rest of this scope, and an explicit `drop` of the pinned reference would not touch it.
    //
    // **Biased, and the order is the decision.** A completed attempt keeps its outcome whatever
    // else happened; a cancellation outranks a deadline that arrived in the same instant, because
    // not checkpointing a cancelled workflow's step is the safer error — a timeout is recorded and
    // replays as a failure, while a cancellation records nothing and runs again on resume. Getting
    // the tie wrong this way costs one re-execution; getting it wrong the other way durably fails a
    // workflow that was never given a chance. Python states the same rule for the same reason.
    tokio::select! {
        biased;
        outcome = attempt => outcome,
        () = cancelled => {
            token.cancel();
            tracing::debug!(
                step_name = name,
                "the workflow was cancelled elsewhere; the step is preempted and records nothing"
            );
            Err(Error::WorkflowCancelled { workflow_id: ctx.workflow_id().to_owned() })
        }
        () = deadline => {
            // Cancelled before this arm returns, and so before the losing future is dropped at the
            // end of the `select!`. A body that could only learn from the token — work on another
            // task, or on a blocking thread — would otherwise never learn at all.
            token.cancel();
            let timeout = options.timeout.expect("the deadline arm only fires with a timeout set");
            tracing::debug!(
                step_name = name,
                timeout_ms = timeout.as_millis(),
                "the step attempt exceeded its timeout and was stopped"
            );
            Err(Error::StepTimeout { step: name.to_owned(), timeout })
        }
    }
}

/// Completes once this workflow is observed `CANCELLED`, and never otherwise.
///
/// **Reuses `await_workflow_result` rather than adding a status channel**, which is the whole of
/// what preemption needs from the database. That method already polls the status row, already
/// reports a cancellation as a *value* rather than an error, already releases its connection
/// between polls, and already runs under the polling-concurrency cap that stops a fan-out of
/// waiters from starving the control plane. Python's poller is the same loop over
/// `get_workflow_status`, with an interval it hardcodes and this one takes from config.
///
/// A terminal outcome that is *not* a cancellation means another executor wrote this workflow's
/// result while this process was still running it. There is nothing to preempt for — the step's
/// own outcome is no longer wanted either way — so this parks rather than reporting a cancellation
/// that did not happen, and lets the attempt finish on its own terms.
async fn observe_cancellation(ctx: &Ctx) {
    let executor = ctx.executor();
    let interval = executor.outcome_poll_interval();
    match executor
        .sysdb()
        // This workflow's own row, which it is running out of, so an absence is a delete.
        .await_workflow_result(ctx.workflow_id(), interval, true)
        .await
    {
        Ok(AwaitedOutcome::Cancelled) => (),
        // Terminal but not a cancellation: another executor wrote this workflow's outcome while
        // this process was still running it. There is nothing to preempt for, so park rather than
        // completing — completing *is* the cancellation signal, and reporting one that did not
        // happen is worse than not watching.
        Ok(_) => std::future::pending().await,
        // **Not retried here, deliberately.** `await_workflow_result` polls in a loop with
        // `with_retry` inside it, and that policy has no attempt limit for transient or connection
        // failures — a database that is merely unreachable never reaches this arm, because the
        // wait blocks until it comes back. What does reach it is the class `sysdb::retry` returns
        // immediately: `Permanent` and non-backend errors, which will fail again identically. So
        // this parks too, and the step finishes on its own terms — the same outcome as not having
        // asked for preemption. Looping here would re-decide a question the layer below owns, and
        // would turn one unparseable status row into a warning every interval for the life of the
        // step.
        Err(error) => {
            tracing::warn!(
                workflow_id = ctx.workflow_id(),
                %error,
                "could not read status for a preemptible step; it will not be preempted"
            );
            std::future::pending().await
        }
    }
}

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

    /// The growth curve, and that the cap is a cap.
    #[test]
    fn backoff_grows_geometrically_and_stops_at_the_cap() {
        let options = StepOptions::<EngineOnly> {
            interval: Duration::from_secs(1),
            backoff_rate: 2.0,
            max_interval: Duration::from_secs(10),
            ..Default::default()
        };
        // The exponent is the number of failures *before* this wait, so the first is un-multiplied.
        assert_eq!(options.backoff(0), Duration::from_secs(1));
        assert_eq!(options.backoff(1), Duration::from_secs(2));
        assert_eq!(options.backoff(2), Duration::from_secs(4));
        assert_eq!(options.backoff(3), Duration::from_secs(8));
        assert_eq!(options.backoff(4), Duration::from_secs(10), "capped");
        assert_eq!(options.backoff(40), Duration::from_secs(10), "still capped");
    }

    /// A rate that cannot produce a duration yields the cap rather than a panic.
    ///
    /// `Duration::from_secs_f64` panics on a negative or non-finite value **and on a finite one
    /// past `u64::MAX` seconds**, and a step retry is the worst place to discover any of the three:
    /// it would take the process down on a config typo, mid-workflow. `1e10` is the case a clamp
    /// applied to the `Duration` rather than to the `f64` misses — `1e20` is finite, positive, and
    /// still unrepresentable.
    #[test]
    fn a_nonsensical_backoff_rate_yields_the_cap() {
        let cap = Duration::from_secs(30);
        // An odd exponent, so a negative rate is still negative by the time it is measured.
        for rate in [-2.0, f64::NAN, f64::INFINITY, 1e10, f64::MAX] {
            let options = StepOptions::<EngineOnly> {
                interval: Duration::from_secs(1),
                backoff_rate: rate,
                max_interval: cap,
                ..Default::default()
            };
            assert_eq!(options.backoff(3), cap, "rate {rate}");
        }
        // The overflow the guard alone does not catch: `1e10^2` is finite and positive, and
        // `1e20` seconds is past `u64::MAX`.
        let overflows = StepOptions::<EngineOnly> {
            interval: Duration::from_secs(1),
            backoff_rate: 1e10,
            max_interval: cap,
            ..Default::default()
        };
        assert_eq!(overflows.backoff(2), cap);
    }

    /// The default is the three-way majority, and it does not retry.
    #[test]
    fn the_defaults_match_python_typescript_and_java() {
        let options = StepOptions::<EngineOnly>::default();
        assert_eq!(options.max_attempts, 1, "a plain step does not retry");
        assert_eq!(options.interval, Duration::from_secs(1));
        assert_eq!(options.backoff_rate, 2.0);
        assert_eq!(options.max_interval, Duration::from_secs(3600));
    }
    use std::sync::Arc;
    use std::sync::atomic::{AtomicU32, Ordering};

    use crate::sysdb::types::{NewWorkflow, StepRecord, Submission};
    use crate::{Config, DBOS};

    /// A launched instance and a workflow row to hang steps off.
    ///
    /// The row has to exist because a step re-stamps its workflow's executor id, and there is no
    /// `run` here on purpose: entering the same workflow id under two contexts is precisely what a
    /// replay is, and it is the cheapest way to produce one without going through recovery.
    async fn workflow(id: &str) -> (DBOS, dbos_test_support::TestDatabase) {
        let db = dbos_test_support::test_database().await;
        let dbos = DBOS::new(Config {
            migrate: false,
            app_version: Some("1.0.0".to_owned()),
            ..Config::new("step-test", db.url())
        });
        dbos.launch().await.expect("launch failed");
        dbos.executor("test")
            .expect("launched")
            .sysdb()
            .init_workflow(&NewWorkflow::new(id), None, Submission::Fresh, None)
            .await
            .expect("could not create the workflow row");
        (dbos, db)
    }

    fn ctx(dbos: &DBOS, id: &str) -> Ctx {
        Ctx::new(dbos.executor("test").expect("launched"), id, None)
    }

    /// The step rows a workflow recorded, in id order.
    async fn steps(dbos: &DBOS, id: &str) -> Vec<StepRecord> {
        dbos.executor("test")
            .expect("launched")
            .sysdb()
            .list_workflow_steps(id, true, None, None, None)
            .await
            .expect("read failed")
    }

    /// What most of these tests actually assert on: which ids were claimed, and by what.
    ///
    /// Split from [`steps`] rather than folded into it because a test that also reads a row's
    /// output needs the rows themselves, and reading them twice to see both halves is worse than
    /// projecting once here.
    fn ids(steps: &[StepRecord]) -> Vec<(i32, &str)> {
        steps
            .iter()
            .map(|step| (step.step_id, step.step_name.as_str()))
            .collect()
    }

    #[tokio::test]
    async fn a_replayed_step_returns_its_recorded_result_without_entering_the_body() {
        let (dbos, _db) = workflow("wf-replay").await;
        let entered = AtomicU32::new(0);
        // Borrows `entered`, so the closure is `Copy` and both runs can use the same body.
        let body = || async {
            entered.fetch_add(1, Ordering::SeqCst);
            Ok::<_, crate::Error>(7u32)
        };

        // First run: the body executes and the result is recorded.
        let first = Ctx::scope(ctx(&dbos, "wf-replay"), async {
            step("compute", body).await
        })
        .await;
        assert_eq!(first.unwrap(), 7);
        assert_eq!(entered.load(Ordering::SeqCst), 1);

        // Replay: a fresh context over the same workflow id, step ids starting again from zero.
        let again = Ctx::scope(ctx(&dbos, "wf-replay"), async {
            step("compute", body).await
        })
        .await;
        assert_eq!(again.unwrap(), 7, "the recorded result");
        assert_eq!(
            entered.load(Ordering::SeqCst),
            1,
            "the body must not run a second time: that is the whole of durable execution"
        );

        dbos.shutdown().await;
    }

    #[derive(Debug, thiserror::Error, serde::Serialize, serde::Deserialize, PartialEq)]
    #[error("the card was declined after {attempts} attempts")]
    struct CardDeclined {
        attempts: u32,
    }

    #[tokio::test]
    async fn a_replayed_step_that_failed_reports_what_it_recorded() {
        let (dbos, _db) = workflow("wf-failed-step").await;

        let failing = || async { Err::<(), _>(CardDeclined { attempts: 3 }.into()) };

        let first = Ctx::scope(ctx(&dbos, "wf-failed-step"), async {
            step("boom", failing).await
        })
        .await;
        let first = first.unwrap_err();
        let Error::Application(error) = &first else {
            panic!("expected an application error, got {first}")
        };
        assert_eq!(error, &CardDeclined { attempts: 3 });

        // Replay gives back the *same error*, decoded, rather than a sentence about it — the same
        // fidelity a successful step's output gets, fields and all.
        let again = Ctx::scope(ctx(&dbos, "wf-failed-step"), async {
            step("boom", failing).await
        })
        .await;
        let again = again.unwrap_err();
        let Error::Application(replayed) = &again else {
            panic!("expected an application error, got {again}")
        };
        assert_eq!(replayed, error, "the failure survives the round trip whole");
        assert_eq!(again.to_string(), first.to_string());

        dbos.shutdown().await;
    }

    /// A database failure is not the step's result either, so nothing is checkpointed.
    ///
    /// The step did not fail; the database did. Recording a failure here would replay as
    /// *permanently* failed, and the next attempt would never re-enter a body that never ran.
    #[tokio::test]
    async fn a_database_failure_is_not_the_steps_result() {
        use crate::sysdb::{BackendError, BackendErrorKind};

        let (dbos, _db) = workflow("wf-blip").await;

        let failed = Ctx::scope(ctx(&dbos, "wf-blip"), async {
            step("charge", || async {
                Err::<(), crate::Error>(Error::SystemDatabase(crate::sysdb::Error::Backend(
                    BackendError {
                        message: "connection reset by peer".to_owned(),
                        sqlstate: None,
                        kind: BackendErrorKind::Connection,
                    },
                )))
            })
            .await
        })
        .await
        .unwrap_err();
        assert!(matches!(failed, Error::SystemDatabase(_)), "{failed}");

        let steps = steps(&dbos, "wf-blip").await;
        assert!(
            steps.is_empty(),
            "a database failure must not be checkpointed as the step's outcome: {steps:?}"
        );

        dbos.shutdown().await;
    }

    /// The one case full fidelity cannot cover: a row this SDK did not write.
    ///
    /// A portable invocation leaves an error encoded in somebody else's shape, and no type
    /// parameter reconstructs a type that was never Rust's. The replay degrades to the message
    /// rather than failing to decode, so the workflow reports what happened instead of reporting
    /// that it could not read what happened.
    #[tokio::test]
    async fn a_step_recorded_in_a_foreign_shape_degrades_to_its_message() {
        let (dbos, _db) = workflow("wf-foreign").await;

        dbos.executor("test")
            .expect("launched")
            .sysdb()
            .record_step(
                "wf-foreign",
                0,
                "charge",
                Outcome::Error(r#"{"pythonModule":"app.errors","type":"CardDeclined"}"#),
                Some("pickle"),
                None,
            )
            .await
            .expect("could not record the step");

        let replayed = Ctx::scope(ctx(&dbos, "wf-foreign"), async {
            step("charge", || async {
                Err::<(), _>(CardDeclined { attempts: 1 }.into())
            })
            .await
        })
        .await
        .unwrap_err();

        match &replayed {
            Error::StepFailed { step, message } => {
                assert_eq!(step, "charge");
                assert!(message.contains("CardDeclined"), "{message}");
            }
            other => panic!("expected a degraded step failure, got {other:?}"),
        }

        dbos.shutdown().await;
    }

    /// A failure that is one of *ours* replays as itself, variant and payload intact — not
    /// flattened into an application error carrying its message.
    #[tokio::test]
    async fn a_replayed_step_gives_back_the_same_variant_it_recorded() {
        let (dbos, _db) = workflow("wf-variant").await;

        let failing = || async {
            Err::<(), crate::Error>(Error::NotRegistered {
                key: "checkout/Checkout/eu".to_owned(),
            })
        };
        let first = Ctx::scope(ctx(&dbos, "wf-variant"), async {
            step("boom", failing).await
        })
        .await;
        let first = first.unwrap_err();

        let again = Ctx::scope(ctx(&dbos, "wf-variant"), async {
            step("boom", failing).await
        })
        .await;
        let again = again.unwrap_err();

        match &again {
            Error::NotRegistered { key } => assert_eq!(key, "checkout/Checkout/eu"),
            other => panic!("the variant did not survive the round trip: {other:?}"),
        }
        assert_eq!(again.to_string(), first.to_string());

        dbos.shutdown().await;
    }

    #[tokio::test]
    async fn a_step_whose_name_changed_is_reported_rather_than_silently_matched() {
        let (dbos, _db) = workflow("wf-renamed").await;

        let first = Ctx::scope(ctx(&dbos, "wf-renamed"), async {
            step("old_name", || async { Ok::<_, crate::Error>(1u32) }).await
        })
        .await;
        assert_eq!(first.unwrap(), 1);

        // Step 0 is recorded under a different name, so its result is not this step's result.
        let renamed = Ctx::scope(ctx(&dbos, "wf-renamed"), async {
            step("new_name", || async { Ok::<_, crate::Error>(1u32) }).await
        })
        .await;
        assert!(
            renamed.is_err(),
            "a changed step name must not match a recorded result"
        );

        dbos.shutdown().await;
    }

    #[tokio::test]
    async fn steps_are_recorded_in_order_under_the_ids_they_allocated() {
        let (dbos, _db) = workflow("wf-order").await;

        Ctx::scope(ctx(&dbos, "wf-order"), async {
            step("one", || async { Ok::<_, crate::Error>(1u32) })
                .await
                .unwrap();
            step("two", || async { Ok::<_, crate::Error>(2u32) })
                .await
                .unwrap();
            step("three", || async { Ok::<_, crate::Error>(3u32) })
                .await
                .unwrap();
        })
        .await;

        let steps = steps(&dbos, "wf-order").await;
        assert_eq!(ids(&steps), [(0, "one"), (1, "two"), (2, "three")]);
        assert_eq!(steps[2].output.as_deref(), Some("3"));

        dbos.shutdown().await;
    }

    #[tokio::test]
    async fn a_nested_step_is_a_plain_call_and_allocates_no_id() {
        let (dbos, _db) = workflow("wf-nested").await;

        Ctx::scope(ctx(&dbos, "wf-nested"), async {
            step("outer", || async {
                // A step is a leaf (Go draws the same line in #420). Were this to allocate an id,
                // every step after it would replay against the wrong slot.
                let inner = step("inner", || async { Ok::<_, crate::Error>(1u32) })
                    .await
                    .unwrap();
                Ok::<_, crate::Error>(inner + 1)
            })
            .await
            .unwrap();
            step("after", || async { Ok::<_, crate::Error>(9u32) })
                .await
                .unwrap();
        })
        .await;

        assert_eq!(
            ids(&steps(&dbos, "wf-nested").await),
            [(0, "outer"), (1, "after")],
            "`inner` is not a checkpoint"
        );

        dbos.shutdown().await;
    }

    /// One poll in the right place does not buy a second somewhere else.
    ///
    /// The run checks its placement as its first act and is then past that check, so a step that
    /// is polled once where it belongs and *then* moved would go on running under the context it
    /// captured — recording beneath a step whose own row already covers whatever its body did,
    /// which is the case `StepBuiltElsewhere` exists to refuse. `PendingStep` is `Send` and
    /// `Unpin`, so nothing but the poll can catch that, and it has to catch it every time.
    ///
    /// The first poll is driven by hand because that is the only way to leave a step started and
    /// still holdable; ordinary `await` runs it to completion where it stands.
    #[tokio::test]
    async fn a_step_polled_once_is_refused_when_it_moves() {
        let (dbos, _db) = workflow("wf-relocated").await;
        let c = ctx(&dbos, "wf-relocated");

        let outcome = Ctx::scope(c.clone(), async {
            // The body never finishes, so it does not matter how far one poll gets: whether it
            // suspends in the round trip that asks whether this step has already run or in the
            // body itself, nothing has been written when the step is moved. A body that returns
            // immediately leaves that to scheduling — one poll runs until the first suspension,
            // and on a loaded machine the reply to that round trip can already be waiting, which
            // carries the poll through the body and into the checkpoint's own write. The
            // assertion below would then be about which read happened to be ready rather than
            // about the rule.
            let mut built = step("moved", || async {
                std::future::pending::<()>().await;
                Ok::<_, crate::Error>(1u32)
            });
            let started = std::pin::Pin::new(&mut built)
                .poll(&mut std::task::Context::from_waker(std::task::Waker::noop()));
            assert!(
                started.is_pending(),
                "the first poll should suspend, leaving the run started"
            );
            // Now somewhere its id cannot be honoured: a step body, whose own checkpoint already
            // stands for everything it does.
            c.in_step_scope(CancellationToken::new(), StepStatus::first(0), built)
                .await
        })
        .await;

        assert!(
            matches!(outcome, Err(Error::StepBuiltElsewhere { .. })),
            "a started step is still refused where it moves: {outcome:?}"
        );
        assert!(
            steps(&dbos, "wf-relocated").await.is_empty(),
            "and it records nothing under the workflow it was built in"
        );

        dbos.shutdown().await;
    }

    /// The leaf rule again, for a body that *builds* its inner step rather than awaiting one it
    /// wrote inline.
    ///
    /// `|| step("inner", ..)` hands the inner step back as the attempt's whole future, so the
    /// closure does its work where it is *called* rather than where that future is polled. Called
    /// outside the step's scope it would take an id from the workflow's counter, be refused for
    /// then being polled inside a step body, and spend another id on every retry — so this is the
    /// one shape that says the closure runs where the attempt does.
    #[tokio::test]
    async fn a_body_that_builds_its_step_builds_it_inside_the_attempt() {
        let (dbos, _db) = workflow("wf-built-in-body").await;

        let outer = Ctx::scope(ctx(&dbos, "wf-built-in-body"), async {
            let outer = step("outer", || {
                step("inner", || async { Ok::<_, crate::Error>(1u32) })
            })
            .await;
            step("after", || async { Ok::<_, crate::Error>(9u32) })
                .await
                .unwrap();
            outer
        })
        .await;

        assert_eq!(outer.unwrap(), 1);
        assert_eq!(
            ids(&steps(&dbos, "wf-built-in-body").await),
            [(0, "outer"), (1, "after")],
            "`inner` is a plain call, so it takes no id and `after` keeps the next one"
        );

        dbos.shutdown().await;
    }

    /// A step built in the workflow proper while a sibling's body is in flight is still a step.
    ///
    /// What the per-call-stack marker buys, at the only altitude that shows it: were the answer
    /// kept on the shared workflow state, `held` being mid-body would make `beside` read *itself*
    /// as nested, so `beside` would run plainly and write no row — and every replay of this
    /// workflow would run it again.
    ///
    /// The two notifies are the whole of the interleaving, and there is no sleep in it: `beside` is
    /// built only once `held` says it is inside its body, and `held` returns only once `beside` has
    /// been awaited.
    #[tokio::test]
    async fn a_step_built_while_a_sibling_runs_is_still_checkpointed() {
        let (dbos, _db) = workflow("wf-concurrent").await;
        let inside = Arc::new(tokio::sync::Notify::new());
        let release = Arc::new(tokio::sync::Notify::new());

        Ctx::scope(ctx(&dbos, "wf-concurrent"), async {
            let held = step("held", {
                let inside = Arc::clone(&inside);
                let release = Arc::clone(&release);
                move || {
                    let inside = Arc::clone(&inside);
                    let release = Arc::clone(&release);
                    async move {
                        inside.notify_one();
                        release.notified().await;
                        Ok::<_, crate::Error>(1u32)
                    }
                }
            });
            let beside = async {
                inside.notified().await;
                let value = step("beside", || async { Ok::<_, crate::Error>(2u32) }).await;
                release.notify_one();
                value
            };

            let (held, beside) = tokio::join!(held, beside);
            assert_eq!(held.unwrap(), 1);
            assert_eq!(beside.unwrap(), 2);
        })
        .await;

        assert_eq!(
            ids(&steps(&dbos, "wf-concurrent").await),
            [(0, "held"), (1, "beside")],
            "`beside` is the workflow's own step, not something inside `held`"
        );

        dbos.shutdown().await;
    }

    /// A built step says what it is before it has run, because the run cannot be asked.
    #[tokio::test]
    async fn a_built_step_carries_its_name_and_id() {
        let (dbos, _db) = workflow("wf-identity").await;

        Ctx::scope(ctx(&dbos, "wf-identity"), async {
            let first = step("first", || async { Ok::<_, crate::Error>(1u32) });
            let second = step("second", || async { Ok::<_, crate::Error>(2u32) });
            assert_eq!((first.name(), first.step_id()), ("first", Some(0)));
            assert_eq!((second.name(), second.step_id()), ("second", Some(1)));
            // Awaiting them is not what numbered them, which is the point.
            assert_eq!(second.await.unwrap(), 2);
            assert_eq!(first.await.unwrap(), 1);
        })
        .await;

        dbos.shutdown().await;
    }

    /// Steps driven concurrently take the ids they were **built** with, not the ones they are
    /// polled in.
    ///
    /// The branches are built `a, b, c` and handed to `join!` as `c, b, a`, so the two orders
    /// disagree. Driving them out of build order is the whole assertion: `join!` first-polls its
    /// branches in source order, so an id taken at the first poll would still have numbered them
    /// 0, 1, 2 had they been listed in the order they were built, and this test would have passed
    /// against the bug it exists to catch.
    #[tokio::test]
    async fn steps_driven_out_of_build_order_keep_the_ids_they_were_built_with() {
        let (dbos, _db) = workflow("wf-out-of-order").await;

        Ctx::scope(ctx(&dbos, "wf-out-of-order"), async {
            let a = step("a", || async { Ok::<_, crate::Error>(1u32) });
            let b = step("b", || async { Ok::<_, crate::Error>(2u32) });
            let c = step("c", || async { Ok::<_, crate::Error>(3u32) });
            let (c, b, a) = tokio::join!(c, b, a);
            assert_eq!((a.unwrap(), b.unwrap(), c.unwrap()), (1, 2, 3));
        })
        .await;

        assert_eq!(
            ids(&steps(&dbos, "wf-out-of-order").await),
            [(0, "a"), (1, "b"), (2, "c")],
            "the ids follow source order, which is what a replay builds again"
        );

        dbos.shutdown().await;
    }

    /// Built before the context existed, awaited inside it.
    ///
    /// `Ctx::scope(ctx, step(..))` is how this happens by accident: the argument is evaluated
    /// first, so the step takes no id and would then run unrecorded where the workflow around it
    /// plainly expects a checkpoint.
    #[tokio::test]
    async fn a_step_built_outside_and_polled_inside_is_refused() {
        let (dbos, _db) = workflow("wf-outside-in").await;

        let smuggled = step("early", || async { Ok::<_, crate::Error>(1u32) });
        assert_eq!(
            smuggled.step_id(),
            None,
            "there was no counter to draw from"
        );

        match Ctx::scope(ctx(&dbos, "wf-outside-in"), smuggled).await {
            Err(Error::StepBuiltElsewhere {
                step,
                built,
                polled,
            }) => {
                assert_eq!((step.as_str(), &*built), ("early", "outside a workflow"));
                assert_eq!(&*polled, "in workflow wf-outside-in");
            }
            other => panic!("expected a built-elsewhere refusal, got {other:?}"),
        }

        dbos.shutdown().await;
    }

    /// Built in one workflow, polled in another: the id names a position in the first.
    #[tokio::test]
    async fn a_step_built_in_another_workflow_is_refused() {
        let (dbos, _db) = workflow("wf-donor").await;
        dbos.executor("test")
            .expect("launched")
            .sysdb()
            .init_workflow(&NewWorkflow::new("wf-thief"), None, Submission::Fresh, None)
            .await
            .expect("could not create the second workflow row");

        // Built in the donor and never awaited there, which is what makes its id a claim nobody
        // honours. Yielding the un-awaited step out of the scope is the smuggling under test, so
        // the lint against it is the thing being demonstrated.
        #[allow(clippy::async_yields_async)]
        let smuggled = Ctx::scope(ctx(&dbos, "wf-donor"), async {
            step("borrowed", || async { Ok::<_, crate::Error>(7u32) })
        })
        .await;
        assert_eq!(smuggled.step_id(), Some(0));

        match Ctx::scope(ctx(&dbos, "wf-thief"), smuggled).await {
            Err(Error::StepBuiltElsewhere {
                step,
                built,
                polled,
            }) => {
                assert_eq!(
                    (step.as_str(), &*built),
                    ("borrowed", "in workflow wf-donor")
                );
                assert_eq!(&*polled, "in workflow wf-thief");
            }
            other => panic!("expected a built-elsewhere refusal, got {other:?}"),
        }

        dbos.shutdown().await;
    }

    /// Built inside a step body, carried out, and awaited in the workflow proper.
    ///
    /// By the leaf rule it took no id inside that body, so awaiting it outside would run it
    /// unrecorded at a position the workflow expects to be a checkpoint. Both places share a
    /// workflow id, so only the step marker tells them apart. It is smuggled out through a slot the
    /// body can reach, since a step's output has to serialize and a `PendingStep` does not.
    #[tokio::test]
    async fn a_step_built_inside_a_step_and_awaited_outside_it_is_refused() {
        let (dbos, _db) = workflow("wf-carried-out").await;

        let (out, smuggled) = tokio::sync::oneshot::channel();
        let refused = Ctx::scope(ctx(&dbos, "wf-carried-out"), async {
            // Taken in the closure rather than the async block: the body is `FnMut`, so nothing it
            // captures may escape into a future the closure hands back.
            let mut out = Some(out);
            step("outer", move || {
                let out = out.take();
                async move {
                    if let Some(out) = out {
                        let _ = out.send(step("inner", || async { Ok::<_, crate::Error>(1u32) }));
                    }
                    Ok::<_, crate::Error>(0u32)
                }
            })
            .await
            .expect("the outer step failed");
            smuggled.await.expect("the body sent one out").await
        })
        .await;

        match refused {
            Err(Error::StepBuiltElsewhere {
                step,
                built,
                polled,
            }) => {
                assert_eq!(step, "inner");
                assert_eq!(&*built, "inside a step of workflow wf-carried-out");
                assert_eq!(&*polled, "in workflow wf-carried-out");
            }
            other => panic!("expected a built-elsewhere refusal, got {other:?}"),
        }

        dbos.shutdown().await;
    }

    /// The same rule the other way: built in the workflow proper, carried *into* a step body.
    ///
    /// Its checkpoint would sit beneath a step whose own row already stands for whatever its body
    /// did, which is the leaf rule read from the other side.
    #[tokio::test]
    async fn a_step_carried_into_a_step_body_is_refused() {
        let (dbos, _db) = workflow("wf-carried-in").await;

        let refused = Ctx::scope(ctx(&dbos, "wf-carried-in"), async {
            let claimed = step("claimed", || async { Ok::<_, crate::Error>(1u32) });
            assert_eq!(claimed.step_id(), Some(0), "claimed at a step boundary");

            let mut carried = Some(claimed);
            step("outer", move || {
                let taken = carried.take();
                async move {
                    match taken {
                        Some(inner) => inner.await,
                        None => Ok(0u32),
                    }
                }
            })
            .await
        })
        .await;

        match refused {
            Err(Error::StepBuiltElsewhere {
                step,
                built,
                polled,
            }) => {
                assert_eq!(step, "claimed");
                assert_eq!(&*built, "in workflow wf-carried-in");
                assert_eq!(&*polled, "inside a step of workflow wf-carried-in");
            }
            other => panic!("expected a built-elsewhere refusal, got {other:?}"),
        }

        dbos.shutdown().await;
    }

    #[tokio::test]
    async fn a_step_outside_a_workflow_runs_plainly() {
        let entered = Arc::new(AtomicU32::new(0));
        let value = step("standalone", || {
            let entered = Arc::clone(&entered);
            async move {
                entered.fetch_add(1, Ordering::SeqCst);
                Ok::<_, crate::Error>(3u32)
            }
        })
        .await
        .expect("a step with no workflow around it just runs");

        assert_eq!(value, 3);
        assert_eq!(entered.load(Ordering::SeqCst), 1);
        assert!(Ctx::current().is_none());
    }
}