dbos 0.6.0

//! Waiting on several workflows at once: the first to finish, or all of them.
//!
//! A fan-out starts N workflows and then has to learn when they finish. Awaiting each handle in
//! turn already works and costs nothing in wall-clock — the caller takes about as long as the
//! slowest, which [`WorkflowRef::start_with`](crate::WorkflowRef::start_with) documents — but it
//! answers only in the order the caller happened to ask, and it holds N waits open where one would
//! do. These two calls are the other shape: **one query per interval whatever N is**, answering
//! either *which one finished first* or *tell me when they all have*.
//!
//! ```no_run
//! # async fn drain(dbos: &dbos::DBOS, handles: Vec<dbos::WorkflowHandle<u32>>) -> dbos::Result<()> {
//! let mut handles = handles;
//! while !handles.is_empty() {
//!     // The borrow ends with the block, so the set can be narrowed below.
//!     let id = {
//!         let ids: Vec<&str> = handles.iter().map(|h| h.workflow_id()).collect();
//!         dbos.select_workflow(&ids).await?
//!     };
//!     // Removed before it is awaited, so the next pass waits on the rest.
//!     let at = handles.iter().position(|h| h.workflow_id() == id).expect("in the set");
//!     let winner = handles.swap_remove(at);
//!     println!("{id} finished: {:?}", winner.result().await);
//! }
//! # Ok(()) }
//! ```
//!
//! # What "finished" means here
//!
//! **Settled, not succeeded.** A workflow counts as finished once its status leaves `PENDING`,
//! `ENQUEUED` and `DELAYED` — so a cancelled or dead-lettered member ends the wait like any other.
//! The caller asked which race finished first, and a set that has to be re-waited because one
//! member was cancelled is a wait that can hang on a workflow nobody is running any more.
//!
//! Neither call returns an outcome, and that is not an omission. What a workflow *returned* comes
//! back through its own [`WorkflowHandle::result`](crate::WorkflowHandle::result), typed, and a
//! wait over a set has no one type to give. So these answer *when*, and the handle answers *what*.
//!
//! # Ids in, ids out
//!
//! **These return ids, not handles.** Handing back an owned
//! [`WorkflowHandle`](crate::WorkflowHandle) would mean taking the whole set by value and dropping
//! every loser, which is precisely the wrong thing for the loop the call exists for. So the answer
//! is the winner's **id** — the identity the handle carried anyway, and the same thing the
//! checkpoint stores, so nothing is projected on the way out and re-derived on replay.
//!
//! A position in the slice would be the only positional return in this crate (`cancel_all` and
//! its neighbours take ids and give back ids), is meaningful only against the exact slice it came
//! from, and would force a set with no repeats — all to skip a
//! [`position`](std::iter::Iterator::position) lookup in the drain loop above.
//!
//! `join_workflows` returns nothing: there is nothing to hand back that the caller did not already
//! have.
//!
//! **These take ids, not handles.** Every bulk call here that only names workflows takes
//! `&[&str]` — [`cancel_all`](crate::DBOS::cancel_all), [`delete_all`](crate::DBOS::delete_all),
//! [`resume_all`](crate::DBOS::resume_all) — and a wait is not different enough to spell its
//! argument a second way. It also makes the heterogeneous case expressible: a fan-out whose members
//! return different types has no common `WorkflowHandle<R, E>` to put in a slice.
//!
//! **Both waits are here.** A fan-out wants both questions — which member finished first, and
//! when have they all — and having one without the other would leave the natural pair half-built.
//!
//! **They are named for the concurrency primitives.** `select_workflow` and `join_workflows` name
//! the *shape* a Rust caller already knows it by, from `tokio::select!` and `tokio::join!` —
//! one branch of several wins, or every branch is joined. That is the name the macros over them
//! take, and a function and its macro that do the same thing under two different names is a seam to
//! learn for nothing.
//!
//! **The recorded step name follows the call**, `DBOS.selectWorkflow`: a step listing should
//! name the call the caller wrote. The all-wait records no step at all, so it needs no name.
//!
//! # Called from inside a workflow
//!
//! **`select_workflow` is checkpointed as a step of the calling workflow**, under
//! [`SELECT_WORKFLOW`](crate::sysdb::types::step_names::SELECT_WORKFLOW), because it records a
//! *choice*. A replay that raced again could see a different member finish first and take a
//! different branch, which would make the workflow nondeterministic in the one way a workflow may
//! never be. So the winner's id is the step's output, and a replay hands back the same id without
//! waiting.
//!
//! **`join_workflows` is checkpointed nowhere, on any of its three surfaces.** It decides nothing:
//! every member has settled by the time it returns, in whatever order, so there is no choice for a
//! step to pin. A step recording only that the wait happened is not worth a step id — for a reason
//! stronger than the row being cheap:
//!
//! - **Whatever it could pin is already pinned harder by what comes after it.** An all-wait is
//!   written to be followed by [`result`](crate::WorkflowHandle::result) on each handle, and each
//!   of *those* is a step recording the outcome the workflow branches on. A replay reads those
//!   rows back and never looks at the children at all, so a recorded wait in front of them is a
//!   write no replay path reads.
//! - **A wait whose results are never read has nothing worth pinning either.** The recorded fact
//!   is "they had all settled", and the wait re-derives it by asking once — every member is
//!   settled already, so the poll answers on its first pass.
//! - **The step id is the expensive part, not the row.** An id is a position in this workflow's
//!   counter, and every call that takes one has to be written the same way on every execution or
//!   each later step lands on a slot it did not record. A call that pins no decision should not be
//!   spending one.
//!
//! **What that costs is one narrow case, and it is the honest reading of the call.** A member
//! *resumed* after it settled — a cancelled workflow put back on a queue — is waited for again by
//! a replay, where a recorded step would have returned at once. The wait answers a question about
//! the set *now*, and a `select_workflow` beside it still pins every choice made on the strength
//! of one.
//!
//! **A replay of a first-wait checks what its payload lets it check, and no more.** It can ask
//! whether the recorded winner is still in the set, because the winner is what it recorded anyway.
//! Step *inputs* are not checkpointed anywhere — the step row has no column for them — so a replay
//! whose arguments changed reads back the answer to the question it asked the first time, exactly
//! as a [`sleep`](crate::sleep()) whose duration changed keeps the deadline it recorded.
//!
//! Outside a workflow the first-wait is not checkpointed either, which is the operator's and the
//! client's case; inside a *step* it is plain too, by the leaf rule every id-allocating call in
//! this crate follows. [`StepPlacement`] owns those rules and the argument for each. The all-wait
//! is the same call in all of these places.
//!
//! # Three surfaces, split by where the caller stands
//!
//! Exactly as [`event`](crate::event) splits its reader, and for the same reason rather than for
//! symmetry. The free [`select_workflow`](fn@select_workflow) and
//! [`join_workflows`](fn@join_workflows) are what a **workflow body** calls: they take the executor
//! from the ambient context, so a workflow that waits needs no [`DBOS`] handle — and a registered
//! closure that captured one would be stored inside the very `Arc` it holds a strong reference to,
//! keeping the instance, its executor and its pool alive for the life of the process.
//! [`DBOS::select_workflow`] and [`DBOS::join_workflows`] are for code **outside** a workflow,
//! where there is nothing ambient to take an executor from;
//! [`Client::select_workflow`](crate::Client::select_workflow) and
//! [`Client::join_workflows`](crate::Client::join_workflows) are that same caller from outside the
//! application altogether.
//!
//! # Two macros over the same two calls
//!
//! [`select_workflow!`](macro@crate::select_workflow) and
//! [`join_workflows!`](macro@crate::join_workflows) are the same two waits taking **handles** and
//! giving back **typed results**, which is what the id-taking calls cannot do: a set of ids has no
//! one type, where a fixed list of handles has one per branch. **Each shares its name with the
//! function it wraps**, which costs nothing — macros live in their own namespace, and
//! `futures::join!` sits beside `futures::future::join` on the same terms — and says the true
//! thing: they are one operation in two spellings, taking ids or taking handles. Neither macro
//! records anything of its own; each expands to the wait beside it plus
//! [`result`](crate::WorkflowHandle::result) on the handles it needs, so the checkpoint that pins
//! the choice is the [`select_workflow`](fn@select_workflow) the macro already made.
//!
//! They are macros because a select's answer is a *sum* over branches that may differ in type, and
//! Rust has no anonymous sum type to return; a function would have to give back nested `Either`s,
//! or force one type on every branch. That is why `select!` is a macro everywhere it exists. Both
//! are `macro_rules!`, exported from this crate rather than a companion one, as `tokio::select!`
//! and `tokio::join!` are.
//!
//! Both take an optional instance before a semicolon — `select_workflow!(dbos; ..)` — which is the
//! three-surface split above spelled at the call site, because a macro has no ambient context of
//! its own to consult. The expansion names the *method* rather than a type, so a [`DBOS`] and a
//! [`Client`](crate::Client) are the same two lines.
//!
//! The id-taking calls are not superseded by any of this. A macro is fixed-arity and needs the
//! handles, so a set built in a loop, a set of ids read from somewhere else, and a
//! [`Client`](crate::Client) that never had handles are all still theirs.
//!
//! # Bounding the wait
//!
//! Neither takes a timeout, for the reason
//! [`await_workflow_result`](crate::sysdb::SystemDatabase::await_workflow_result) gives at length:
//! `tokio::time::timeout` around the future, or dropping it, ends the poll: the language already
//! supplies the bound, so the call needs no parameter for it.

use std::sync::Arc;

use crate::checkpoint::{PendingStep, StepPlacement, StepPosition, revive};
use crate::connection::Connection;
use crate::error::{Error, Result};
use crate::instance::DBOS;
use crate::serialization::{decode_tagged, encode};
use crate::sysdb::types::{Outcome, Timestamp, step_names};

/// Waits until one of these workflows finishes, and reports **which**.
///
/// The waiter for a workflow body: it takes the executor from the ambient context, so a workflow
/// that waits on a fan-out needs no [`DBOS`] handle and its registered closure captures nothing.
/// That is what keeps the registry free of strong references back to the instance holding it, and
/// it is why [`get_event`](crate::get_event) is a free function too.
///
/// The wait is checkpointed as a step, so a replay returns the same winner instead of racing again.
/// **Its step id is taken here, at the call, not at the first poll** — see [`PendingStep`] — so a
/// wait built beside a step and driven with it by `tokio::join!` takes the same slot on every
/// execution, whatever it was passed: an empty set is placed like any other and its refusal
/// recorded as the step's own outcome, so a replay refuses again. The error is the *workflow's*
/// channel, like [`step`](crate::step())'s, so `?` needs no conversion. From inside a *step* it
/// waits plainly with no checkpoint, the step's own checkpoint standing for everything its body
/// did.
///
/// ```no_run
/// # async fn fan_out(child: dbos::WorkflowRef<u32, u32>) -> dbos::Result<u32> {
/// let mut handles = Vec::new();
/// for n in 0..3 {
///     handles.push(child.start(n).await?);
/// }
/// let id = {
///     let ids: Vec<&str> = handles.iter().map(|h| h.workflow_id()).collect();
///     dbos::select_workflow(&ids).await?
/// };
/// let at = handles.iter().position(|h| h.workflow_id() == id).expect("in the set");
/// handles.swap_remove(at).result().await
/// # }
/// ```
///
/// Outside a workflow there is no context to read, so this is [`Error::NotInWorkflow`]. That is
/// where [`DBOS::select_workflow`] is the call.
///
/// **This rather than a [`select_step!`](crate::select_step) over the handles' awaits**, wherever
/// the branches are all workflows. Both are durable — an await is a step like any other — but this
/// settles the whole set with one query per poll interval, where N raced awaits are N pollers
/// asking separately. `select_step!` is for a race whose branches are not all of one kind.
pub fn select_workflow<'a, E: crate::DurableError + 'a>(
    workflow_ids: &'a [&'a str],
) -> PendingStep<'a, String, E> {
    let built = StepPlacement::ambient_connection("select_workflow")
        .and_then(|conn| Connection::placed(&conn, "select_workflow"));
    Connection::pending_select_workflow(built, workflow_ids)
}

/// Waits until every one of these workflows has finished.
///
/// The all-form of the free [`select_workflow`](fn@select_workflow), and the waiter a fan-out that
/// needs every answer reaches for. Same context rules and same error channel — but **nothing is
/// checkpointed and no step id is taken**, here or on any other surface, for the reasons the
/// module doc gives: an all-wait decides nothing, and the
/// [`result`](crate::WorkflowHandle::result) calls it exists to precede record the outcomes a
/// replay actually reads.
///
/// So this is an ordinary `async fn` where the first-wait is a [`PendingStep`], and being inside a
/// workflow is what it needs only to find an executor to wait through — which it looks for when
/// the future is first polled, since it has nothing to claim before then. Outside a workflow that
/// is [`Error::NotInWorkflow`], and [`DBOS::join_workflows`] is the call.
///
/// **An empty set is a satisfied wait**, answered at once, where an empty
/// [`select_workflow`](fn@select_workflow) is refused. Answering early is safe only because this
/// takes no step id: a set computed from state could be empty on one execution and not the next,
/// and a call that took an id in one and none in the other would shift every later step.
///
/// **A replay asks the set again, rather than returning from anything it recorded.** That is the
/// point of taking no step id, and it is what makes the members' *current* state the answer. The
/// case where it shows is a member deleted between one execution and the next: it never settles, so
/// the replayed join keeps waiting, where [`result`](crate::WorkflowHandle::result) on the same
/// handle would have read the outcome back out of the parent's own row.
///
/// Neither is the defined behaviour of a deleted workflow. Removing one that a parent is still
/// replaying through is collecting what is not garbage — the parent's recorded outcomes make some
/// paths survive it, this one does not, and the difference is not a contract either way. A fan-out
/// that must tolerate it should await its handles rather than join them first.
///
/// ```no_run
/// # async fn fan_out(child: dbos::WorkflowRef<u32, u32>) -> dbos::Result<u32> {
/// let mut handles = Vec::new();
/// for n in 0..3 {
///     handles.push(child.start(n).await?);
/// }
/// let ids: Vec<&str> = handles.iter().map(|h| h.workflow_id()).collect();
/// dbos::join_workflows(&ids).await?;
/// // Every handle now resolves without waiting.
/// let mut total = 0;
/// for handle in handles {
///     total += handle.result().await?;
/// }
/// # Ok(total) }
/// ```
pub async fn join_workflows<E: crate::DurableError>(workflow_ids: &[&str]) -> crate::Result<(), E> {
    let conn = StepPlacement::ambient_connection("join_workflows").map_err(Error::lift)?;
    Connection::join_workflows(&conn, workflow_ids)
        .await
        .map_err(Error::lift)
}

/// Races these workflow handles and runs the arm belonging to the one that finishes first.
///
/// The macro form of [`select_workflow`](fn@select_workflow). Where that answers with the winner's
/// **id**, this hands the winning arm the winner's **typed result** — so a fan-out over children
/// that return different types is one expression rather than an id, a lookup against the set, and a
/// cast.
///
/// ```no_run
/// # async fn race(
/// #     counter: dbos::WorkflowRef<u32, u32>,
/// #     namer: dbos::WorkflowRef<(), String>,
/// # ) -> dbos::Result<String> {
/// let count = counter.start(3).await?;
/// let name = namer.start(()).await?;
/// dbos::select_workflow! {
///     n = count => format!("the counter won with {}", n?),
///     s = name  => format!("the namer won with {}", s?),
/// }
/// # }
/// ```
///
/// # Inside a workflow
///
/// **Nothing new is recorded.** The expansion is one [`select_workflow`](fn@select_workflow) over
/// the handles' ids — the checkpoint that already makes this choice survive a replay — and then
/// [`result`](crate::WorkflowHandle::result) on the winner alone. A replayed body reads the same
/// winner back out of that checkpoint and awaits the same handle, so the arm taken the second time
/// is the arm taken the first.
///
/// **Only the winner is awaited**, and that is what keeps the step ids stable rather than being an
/// optimisation: a loser's `result` is called on neither the run nor the replay, so no id is taken
/// on one and not the other. Dropping the losing handles does not stop those workflows — a
/// workflow runs on whether or not anything is watching it, which is the difference between racing
/// workflows and racing steps.
///
/// # Outside a workflow
///
/// An instance before a semicolon is the other form, and it is the [`DBOS::select_workflow`]
/// surface where the bare form is the free [`select_workflow`](fn@select_workflow) one — the same
/// split the calls themselves make, spelled at the call site because a macro has no ambient context
/// of its own to consult:
///
/// ```no_run
/// # async fn race(dbos: &dbos::DBOS, a: dbos::WorkflowHandle<u32>, b: dbos::WorkflowHandle<u32>)
/// # -> dbos::Result<u32> {
/// dbos::select_workflow! { dbos;
///     first = a => first?,
///     second = b => second?,
/// }
/// # }
/// ```
///
/// Anything with the two waits on it goes there, which today is a [`DBOS`] or a
/// [`Client`](crate::Client) — the macro names the method rather than a type, so a client's
/// undurable wait and an instance's checkpointed one are the same expansion. **Inside a workflow,
/// prefer the bare form**: it needs no handle, for the reason a registered closure that captured a
/// [`DBOS`] would keep the instance alive for the life of the process, and a client's wait inside a
/// workflow body is not checkpointed and so runs again on every replay.
///
/// One wrinkle belongs to this form alone: an instance answers in the engine's own channel, so the
/// expansion lifts that error into the caller's, and where nothing else says which channel that is
/// — a test, or any caller that never `?`s the result — it has to be annotated. Inside a workflow
/// the body's own return type settles it and the question does not arise.
///
/// # The shape of an arm
///
/// **A branch is a variable holding a handle, not an arbitrary expression.** The handle is named
/// twice, once for its id and once to consume it, and an expression would be evaluated twice —
/// which for `child.start(n).await?` would start the workflow twice. A handle a body means to race
/// has been bound to a name by the time it gets here in any case.
///
/// **Two branches naming the same workflow are answered by the first**, in the order the arms are
/// written. [`select_workflow`](fn@select_workflow) accepts a repeated id for the same reason: the
/// answer names one workflow however many entries pointed at it.
///
/// The value is a [`Result`], because the wait itself can fail. Each arm binds its own handle's
/// result, so an arm decides for itself whether to `?` it, match it, or report it.
#[macro_export]
macro_rules! select_workflow {
    // The two public forms differ only in which wait they reach for, so both hand the same arms to
    // this one along with the call that produces the winner. Internal rules come first because a
    // rule that fails on a literal backtracks cleanly, where one that fails inside a `pat` or
    // `expr` fragment need not.
    ( @race { $wait:expr } $( $bind:pat = $handle:ident => $arm:expr ),+ ) => {
        'dbos_select: {
            let winner = match $wait {
                ::core::result::Result::Ok(winner) => winner,
                ::core::result::Result::Err(failed) => {
                    break 'dbos_select ::core::result::Result::Err(failed);
                }
            };
            $(
                if $handle.workflow_id() == winner {
                    let $bind = $handle.result().await;
                    break 'dbos_select ::core::result::Result::Ok($arm);
                }
            )+
            // `select_workflow` checks its recorded winner against the set it was given, so an id
            // from outside it has already been reported as `UnexpectedStep` before this is reached.
            ::core::unreachable!(
                "select_workflow answered with {winner}, which is none of the handles it was given"
            )
        }
    };
    ( $( $bind:pat = $handle:ident => $arm:expr ),+ $(,)? ) => {
        $crate::select_workflow!(
            @race { $crate::select_workflow(&[$( $handle.workflow_id() ),+]).await }
            $( $bind = $handle => $arm ),+
        )
    };
    ( $instance:expr ; $( $bind:pat = $handle:ident => $arm:expr ),+ $(,)? ) => {
        // `lift` because an instance and a client answer in the engine's own channel, where the
        // arms speak the application's: the free form is generic over that channel and needs no
        // conversion, and this one is the `map_err(Error::lift)` every other engine-channel call
        // asks of its caller.
        $crate::select_workflow!(
            @race {
                $instance
                    .select_workflow(&[$( $handle.workflow_id() ),+])
                    .await
                    .map_err($crate::Error::lift)
            }
            $( $bind = $handle => $arm ),+
        )
    };
    // Last, so it speaks only for input no other rule claimed. Worth the rule because the two
    // shapes it covers are the two mistakes available here, and rustc's own answer to both is
    // "no rules expected this token".
    ( $( $bad:tt )* ) => {
        ::core::compile_error!(
            "select_workflow! takes arms of the form `binding = handle => expression`, where each \
             handle is a variable holding a WorkflowHandle rather than an expression. An instance \
             to wait through goes before a *semicolon*: `select_workflow!(dbos; won = a => won?)`."
        )
    };
}

/// Waits for all of these workflows and returns what each of them returned.
///
/// The macro form of [`join_workflows`](fn@join_workflows), and the reason to reach for it rather
/// than awaiting each handle in turn: one poll loop settles the whole set — **one query per
/// interval whatever N is** — and only then is each result read, so N handles cost one wait and N
/// row reads instead of N waits.
///
/// ```no_run
/// # async fn both(
/// #     counter: dbos::WorkflowRef<u32, u32>,
/// #     namer: dbos::WorkflowRef<(), String>,
/// # ) -> dbos::Result<String> {
/// let count = counter.start(3).await?;
/// let name = namer.start(()).await?;
/// let (count, name) = dbos::join_workflows!(count, name)?;
/// # Ok(format!("{name} counted to {count}")) }
/// ```
///
/// **The results are what get recorded, and the awaits stay sequential.** The expansion is one
/// [`join_workflows`](fn@join_workflows), which records nothing, and then
/// [`result`](crate::WorkflowHandle::result) on each handle in source order — so the whole of what
/// a replay reads back is the N `DBOS.getResult` rows. Sequential is not a concession here: the
/// set is already settled by the time the first result is read, so every one of them is a row
/// read that does not wait — and taking them in source order is what keeps each `DBOS.getResult`
/// on the step id its replay expects.
///
/// **The first failure ends it**, in source order, as `try_join!` does: the tuple holds values
/// rather than results, so a failed child is the value of the whole call. Where each child's own
/// failure matters, [`join_workflows`](fn@join_workflows) followed by a `result` per handle reports
/// all of them.
///
/// An instance before a semicolon is the outside-a-workflow form, exactly as it is for
/// [`select_workflow!`](macro@crate::select_workflow), which sets out when to reach for which:
///
/// ```no_run
/// # async fn both(dbos: &dbos::DBOS, a: dbos::WorkflowHandle<u32>, b: dbos::WorkflowHandle<u32>)
/// # -> dbos::Result<u32> {
/// let (a, b) = dbos::join_workflows!(dbos; a, b)?;
/// # Ok(a + b) }
/// ```
///
/// A branch is a variable holding a handle, for the reason
/// [`select_workflow!`](macro@crate::select_workflow) gives.
///
/// **A comma there is not caught.** `join_workflows!(dbos, a, b)` is a perfectly good list of three
/// idents, so it is read as three handles and fails inside the expansion, where the instance turns
/// out to have no `workflow_id`. That ambiguity is why the separator is a semicolon at all: unlike
/// [`select_workflow!`](macro@crate::select_workflow), whose arms cannot be mistaken for an
/// instance, nothing distinguishes the two shapes here.
#[macro_export]
macro_rules! join_workflows {
    ( @join { $wait:expr } $( $handle:ident ),+ ) => {
        'dbos_join: {
            if let ::core::result::Result::Err(failed) = $wait {
                break 'dbos_join ::core::result::Result::Err(failed);
            }
            ::core::result::Result::Ok((
                $(
                    match $handle.result().await {
                        ::core::result::Result::Ok(value) => value,
                        ::core::result::Result::Err(failed) => {
                            break 'dbos_join ::core::result::Result::Err(failed);
                        }
                    },
                )+
            ))
        }
    };
    ( $( $handle:ident ),+ $(,)? ) => {
        $crate::join_workflows!(
            @join { $crate::join_workflows(&[$( $handle.workflow_id() ),+]).await }
            $( $handle ),+
        )
    };
    ( $instance:expr ; $( $handle:ident ),+ $(,)? ) => {
        $crate::join_workflows!(
            @join {
                $instance
                    .join_workflows(&[$( $handle.workflow_id() ),+])
                    .await
                    .map_err($crate::Error::lift)
            }
            $( $handle ),+
        )
    };
    // The comma form this cannot catch is called out by name: `join_workflows!(dbos, a, b)` is a
    // valid list of three idents, so it matches the bare rule above and fails later, inside the
    // expansion, on an instance that has no `workflow_id`.
    ( $( $bad:tt )* ) => {
        ::core::compile_error!(
            "join_workflows! takes variables holding WorkflowHandles: `join_workflows!(a, b)`. An \
             instance to wait through goes before a *semicolon*: `join_workflows!(dbos; a, b)` — \
             with a comma there, the instance is read as one more handle."
        )
    };
}

impl DBOS {
    /// Waits until one of these workflows finishes, and reports **which**.
    ///
    /// The waiter for code outside a workflow — an operator's tool, or an HTTP handler watching a
    /// batch it kicked off. **Inside a workflow, reach for the free
    /// [`select_workflow`](fn@select_workflow) instead**: it needs no handle, so the closure a
    /// workflow is registered as captures nothing, and a captured [`DBOS`] is a cycle with the
    /// registry that holds the closure. Called from inside one anyway, it behaves as the free
    /// function does, except that it reports in the engine's own error channel — and that a handle
    /// to some *other* instance is [`Error::WrongInstance`], because this takes its executor from
    /// `self` and its step id from the ambient context.
    ///
    /// **"Finishes" means settled, not succeeded**: a workflow counts once its status leaves
    /// `PENDING`, `ENQUEUED` and `DELAYED`, so a cancelled or dead-lettered member ends the wait
    /// like any other. What a workflow *returned* comes back through its own
    /// [`WorkflowHandle::result`](crate::WorkflowHandle::result), typed — this answers *when*, and
    /// the handle answers *what*.
    ///
    /// **The winner's id, not its handle.** Returning an owned
    /// [`WorkflowHandle`](crate::WorkflowHandle) would mean taking the whole set by value and
    /// dropping every loser, which is the wrong thing for the loop this call exists for. The id is
    /// the identity that handle carried, it is what the checkpoint stores, and it is what every
    /// other bulk call in this crate deals in.
    ///
    /// ```no_run
    /// # async fn f(dbos: &dbos::DBOS, a: &str, b: &str) -> dbos::Result<()> {
    /// let first = dbos.select_workflow(&[a, b]).await?;
    /// println!("{first} got there first");
    /// # Ok(()) }
    /// ```
    ///
    /// **A tie is broken arbitrarily.** Two workflows that settle within the same poll interval are
    /// not ordered by anything — the interval is the resolution at which this can observe
    /// finishing. Inside a workflow that costs nothing, because the winner is recorded and a replay
    /// reads it back instead of racing again; outside one, a second call over the same settled set
    /// may answer differently.
    /// [`await_first_workflow_id`](crate::sysdb::SystemDatabase::await_first_workflow_id) sets out
    /// why an `ORDER BY` would make this worse rather than better.
    ///
    /// **Duplicate ids are accepted**, here and in [`join_workflows`](Self::join_workflows). An
    /// answer that is an id has no collision to fear — a set with `a` twice answers `a`, which
    /// names one workflow however many entries pointed at it. A caller-supplied id that
    /// [`start`](crate::WorkflowRef::start) joined to a run already going produces exactly that
    /// set, legitimately, and there is nothing here for it to break.
    ///
    /// **An empty slice is refused** rather than waited on — the one input either wait rejects.
    /// A wait for one of nothing has no answer it could ever give, where a wait for *all* of
    /// nothing is already satisfied.
    ///
    /// **An id naming no workflow is waited for, not reported**, which is what makes a set of ids
    /// usable before every enqueue has committed — and what makes a mistyped id a wait that never
    /// ends. The bound on that is dropping the future.
    pub fn select_workflow<'a>(&self, workflow_ids: &'a [&'a str]) -> PendingStep<'a, String> {
        let built = self
            .executor("select_workflow")
            .and_then(|executor| Connection::placed(executor.connection(), "select_workflow"));
        Connection::pending_select_workflow(built, workflow_ids)
    }

    /// Waits until every one of these workflows has finished.
    ///
    /// The all-form of [`DBOS::select_workflow`], with the same note about the free
    /// [`join_workflows`](fn@join_workflows) being the call a workflow body should make.
    ///
    /// **"Finishes" means settled, not succeeded** — see
    /// [`select_workflow`](Self::select_workflow), which shares this call's definition of it.
    /// Nothing is returned because there is nothing to hand back that the caller did not already
    /// have.
    ///
    /// ```no_run
    /// # async fn f(dbos: &dbos::DBOS, handles: Vec<dbos::WorkflowHandle<u32>>) -> dbos::Result<()> {
    /// let ids: Vec<&str> = handles.iter().map(|h| h.workflow_id()).collect();
    /// dbos.join_workflows(&ids).await?;
    /// // Every handle now resolves without waiting.
    /// # Ok(()) }
    /// ```
    ///
    /// **Duplicates are accepted**, as they are in [`select_workflow`](Self::select_workflow):
    /// settling is a property of an id, so a repeated one is simply satisfied twice. **An empty
    /// slice returns at once** — nothing to wait for is a satisfied wait, where an empty first-wait
    /// has no answer and is refused.
    ///
    /// **Nothing is checkpointed**, unlike [`select_workflow`](Self::select_workflow) — the module
    /// doc argues it. Two consequences follow from that and are worth stating. A member *resumed*
    /// after it settled, a cancelled workflow put back on a queue, is waited for again by a replay
    /// where a recorded step would have returned at once — that is the call answering the question
    /// it was asked, which is about the set now. And a member *deleted* between one execution and
    /// the next never settles, so a replayed join waits where awaiting each handle in turn would
    /// have read back the outcomes the parent already recorded; the free
    /// [`join_workflows`](fn@join_workflows) sets out why, and what to reach for instead.
    ///
    /// Called from inside a workflow it behaves exactly as it does outside one, so an instance
    /// that is not the ambient workflow's is a plain wait here rather than
    /// [`Error::WrongInstance`]: there is no step id for two counters to disagree about, which is
    /// the only thing that error ever guarded.
    pub async fn join_workflows(&self, workflow_ids: &[&str]) -> Result<()> {
        let executor = self.executor("join_workflows")?;
        Connection::join_workflows(executor.connection(), workflow_ids).await
    }
}

/// The same two waits from outside the application.
///
/// Identical to [`DBOS`]'s, with the two differences a client always has: there is no launch
/// check, because a client is connected or it does not exist; and **nothing is checkpointed**,
/// because a client has no step counter of its own to agree with a workflow's. Called from inside
/// a workflow body, a client's wait runs again on replay — the line
/// [`WorkflowHandle::result`](crate::WorkflowHandle::result) already draws for a client's handle
/// awaited there.
///
/// Both waits are here because this is the surface an operator's tool wants most: a client is the
/// caller most likely to hold a set of ids it did not start.
impl crate::Client {
    /// Waits until one of these workflows finishes, and reports which.
    ///
    /// See [`DBOS::select_workflow`]. Nothing is checkpointed, because a client has nothing to
    /// checkpoint against.
    pub async fn select_workflow(&self, workflow_ids: &[&str]) -> Result<String> {
        let (conn, placement) = Connection::placed(self.connection(), "select_workflow")?;
        Connection::select_workflow(&conn, placement, workflow_ids).await
    }

    /// Waits until every one of these workflows has finished.
    ///
    /// See [`DBOS::join_workflows`], which a client reaches by the same body: an all-wait is
    /// checkpointed nowhere, so this surface differs from that one only in where it finds a
    /// connection.
    pub async fn join_workflows(&self, workflow_ids: &[&str]) -> Result<()> {
        Connection::join_workflows(self.connection(), workflow_ids).await
    }
}

/// Names an id set for an error message, without letting a fan-out of thousands *become* the
/// message.
///
/// The first few and a count: enough to see which set is meant, where the whole of a wide one
/// would be a wall of ids that says no more than its first line did.
fn summarize(workflow_ids: &[&str]) -> String {
    const SHOWN: usize = 5;
    if workflow_ids.len() <= SHOWN {
        workflow_ids.join(", ")
    } else {
        format!(
            "{}, and {} more",
            workflow_ids[..SHOWN].join(", "),
            workflow_ids.len() - SHOWN
        )
    }
}

impl Connection {
    /// The connection the first-wait goes through, paired with where that wait stands.
    ///
    /// **The set is not looked at here**, which is the point rather than an omission: which slot a
    /// wait occupies must depend on where it was written, never on what it was passed. A set
    /// computed from state can be empty on one execution and not on the next, and a call that took
    /// an id in one and none in the other shifts every later step of that workflow onto a slot it
    /// did not record. So an empty set is placed like any other and refused by the run, with the
    /// refusal recorded. The all-wait takes no id at all and so has no slot to keep stable.
    ///
    /// A repeated id is not looked at either: the first-wait answers with the id itself, which
    /// names one workflow however many entries pointed at it.
    pub(crate) fn placed(
        conn: &Arc<Self>,
        operation: &'static str,
    ) -> Result<(Arc<Self>, StepPlacement)> {
        let placement = StepPlacement::of(conn, operation)?;
        Ok((Arc::clone(conn), placement))
    }

    /// The first-wait as a [`PendingStep`], for the two surfaces that take a step id.
    ///
    /// The free [`select_workflow`](fn@crate::select_workflow) and [`DBOS::select_workflow`] differ
    /// only in how they reach a connection and in which error channel they answer in, which is the
    /// whole of what `built` carries; a [`Client`](crate::Client)'s stays a plain `async fn`,
    /// because it takes no id and is legitimately driven from anywhere.
    pub(crate) fn pending_select_workflow<'a, E: crate::DurableError + 'a>(
        built: Result<(Arc<Self>, StepPlacement)>,
        workflow_ids: &'a [&'a str],
    ) -> PendingStep<'a, String, E> {
        PendingStep::placed(
            step_names::SELECT_WORKFLOW,
            built,
            move |conn, placement| async move {
                Self::select_workflow(&conn, placement, workflow_ids)
                    .await
                    .map_err(Error::lift)
            },
        )
    }

    /// The first-wait itself, once its slot is decided.
    ///
    /// On the connection because that is what it needs — a poll interval, a serializer for the
    /// recorded winner, and the database — which is what lets a client reach it, and what makes
    /// this the one body all three surfaces run.
    pub(crate) async fn select_workflow(
        self: &Arc<Self>,
        placement: StepPlacement,
        workflow_ids: &[&str],
    ) -> Result<String> {
        // The ids first, so the refusal below has them: `check` answers `Some` only where the
        // placement holds a step, so inside this arm they are known rather than defaulted.
        if let Some(StepPosition {
            workflow_id,
            step_id,
        }) = placement.step_position()
            && let Some(recorded) = placement.check(self, step_names::SELECT_WORKFLOW).await?
        {
            // A recorded failure replays as that failure. The only one this call records is the
            // empty set below, and a replay of it must refuse again rather than look at the set it
            // has now — the same rule that makes a recorded winner stand even when the set changed.
            if let Some(error) = &recorded.error {
                return Err(revive(
                    error,
                    recorded.serialization.as_deref(),
                    self.serializer(),
                    step_names::SELECT_WORKFLOW,
                ));
            }
            let winner: String = decode_tagged(
                recorded.output.as_deref(),
                recorded.serialization.as_deref(),
                self.serializer(),
                "the id that won a select_workflow",
            )?;
            tracing::debug!(
                workflow_id = winner,
                "replaying select_workflow; the same workflow wins again"
            );
            // **The recorded winner has to still be in the set**, which is a determinism check and
            // not bookkeeping: a workflow that changed which ids it waits on has changed what this
            // position of its code means, and handing back a winner it no longer waits on would
            // have it act on an answer to a question it stopped asking. The same check
            // `ChildResultPlacement::check` makes against a recorded child id, for the same reason.
            //
            // **Spelled out because nothing else would catch it.** Answering with the id means
            // nothing here dereferences it against the set, as looking up a winning handle would.
            // It reads no extra state: the winner is the payload this call records anyway.
            if !workflow_ids.contains(&winner.as_str()) {
                // `expected` is what this run is asking for and `recorded` what the row holds,
                // which is the order `Error::UnexpectedStep` prints them in and the order
                // `ChildResultPlacement::check` builds them in.
                return Err(Error::SystemDatabase(crate::sysdb::Error::UnexpectedStep {
                    workflow_id: workflow_id.to_owned(),
                    step_id,
                    expected: format!("a select_workflow over {}", summarize(workflow_ids)),
                    recorded: format!("a select_workflow won by {winner}"),
                }));
            }
            return Ok(winner);
        }

        // Taken before the wait, so a workflow's timeline shows the waiting rather than the
        // instant the answer was written down — the placement every other recorded call in this
        // crate takes its `started_at` from.
        let started_at = Timestamp::now();

        // **A first-wait over nothing has no answer it could ever give**, so it is refused rather
        // than parked forever — and the refusal is *recorded*, because this call took a step id
        // and a step that took one owes its slot an outcome. A replay then reads the refusal back
        // instead of re-deciding it, which is what every other recorded failure does. Outside a
        // workflow there is no slot, so `record` writes nothing and this is simply the error.
        if workflow_ids.is_empty() {
            let refused: Error = Error::InvalidArgument {
                operation: "select_workflow".into(),
                detail: "no workflow ids to wait for".to_owned(),
            };
            let encoded = encode(
                &refused,
                &self.serializer().encoding(),
                "select_workflow error",
            )?;
            placement
                .record(
                    self,
                    step_names::SELECT_WORKFLOW,
                    Outcome::Error(&encoded),
                    started_at,
                )
                .await?;
            return Err(refused);
        }
        let winner = self
            .sysdb()
            .await_first_workflow_id(workflow_ids, self.outcome_poll_interval())
            .await
            .map_err(Error::SystemDatabase)?;

        let encoded = encode(
            &winner,
            &self.serializer().encoding(),
            "the id that won a select_workflow",
        )?;
        placement
            .record(
                self,
                step_names::SELECT_WORKFLOW,
                Outcome::Output(Some(&encoded)),
                started_at,
            )
            .await?;
        Ok(winner)
    }

    /// The all-wait itself — see [`select_workflow`](Self::select_workflow), of which this is the
    /// other half, and which is the one of the two that records anything.
    ///
    /// **Takes no placement, because there is no slot to decide.** All three surfaces run this
    /// same body, where the first-wait's has one shape inside a workflow and another outside it;
    /// the module doc sets out why an all-wait pins nothing worth a step id.
    ///
    /// On the connection because that is what it needs — a poll interval and the database —
    /// which is what lets a client reach it.
    pub(crate) async fn join_workflows(self: &Arc<Self>, workflow_ids: &[&str]) -> Result<()> {
        self.sysdb()
            .await_workflow_ids(workflow_ids, self.outcome_poll_interval())
            .await
            .map_err(Error::SystemDatabase)
    }
}