dbos 0.6.0

//! Child workflows: a workflow started from inside another one, against real databases.

use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::Duration;

use dbos::sysdb::SystemDatabase;
use dbos::sysdb::postgres::{PostgresSystemDatabase, Settings};
use dbos::sysdb::types::{Outcome, WorkflowStatus};
use dbos::{
    Config, DBOS, Debouncer, DuplicationPolicy, Error, QueueConflict, QueueOptions, QueueRequest,
    RunOptions, StartOptions, Timeout, WorkflowIdReusePolicy,
};

use dbos_test_support::{Backend, TestDatabase, owner_token, test_database};

/// The version each instance in this file launches with, derived from its application name.
///
/// DBOS computes none, so a launch without one fails, and it has to be stable across a relaunch of
/// the same application or the relaunch would recover nothing. It cannot simply be a shared
/// constant, though: `application_versions` carries a global `UNIQUE (version_name)`, so two
/// differently-named applications sharing one database cannot both register `1.0.0` — see
/// `resolve_owning_application` in the system database, and
/// `lifecycle::a_launch_that_fails_after_connecting_closes_the_database`, which contests a version
/// on purpose. Keying on the name gives both properties at once.
fn app_version(app_name: &str) -> String {
    format!("{app_name}-1.0.0")
}

fn config(app_name: &str, db: &TestDatabase) -> Config {
    Config {
        migrate: false,
        app_version: Some(app_version(app_name)),
        ..Config::new(app_name, db.url())
    }
}

async fn reader(db: &TestDatabase) -> PostgresSystemDatabase {
    PostgresSystemDatabase::from_pool(db.pool().await, &Settings::default())
}

/// The child's id is `{parent}-{step_id}`, its row records the parent, and the parent records the
/// start as a step.
///
/// Zero-based, like the step ids the name is built from: a parent's first child is `parent-0`.
///
/// **A `run` costs two step ids** — the start and the await — so consecutive children are `-0` and
/// `-2` rather than `-0` and `-1`: one counter, and both halves take from it.
#[tokio::test]
async fn a_child_is_named_for_its_parent_and_the_step_that_started_it() {
    let db = test_database().await;
    let dbos = DBOS::new(config("child-id-app", &db));
    let child = dbos
        .register_workflow("child", |n: u32| async move { Ok::<u32, Error>(n * 2) })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", move |()| {
            let child = child.clone();
            async move {
                let first = child.run(21).await?;
                let second = child.run(50).await?;
                Ok::<u32, Error>(first + second)
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let id = "the-parent";
    let total = parent
        .run_with(
            (),
            RunOptions {
                workflow_id: Some(id),
                ..RunOptions::default()
            },
        )
        .await
        .expect("the parent failed");
    assert_eq!(total, 142, "both children ran");

    let reader = reader(&db).await;
    for (step_id, expected) in [(0, "the-parent-0"), (2, "the-parent-2")] {
        let row = reader
            .get_workflow(expected)
            .await
            .expect("read failed")
            .unwrap_or_else(|| panic!("no row for {expected}"));
        assert_eq!(row.status, WorkflowStatus::Success);
        assert_eq!(
            row.parent_workflow_id.as_deref(),
            Some(id),
            "the child's row points back at its parent"
        );
        let steps = reader
            .list_workflow_steps(id, false, None, None, None)
            .await
            .expect("read failed");
        let step = steps
            .iter()
            .find(|step| step.step_id == step_id)
            .unwrap_or_else(|| panic!("no step {step_id} on the parent"));
        assert_eq!(step.step_name, "child");
        assert_eq!(
            step.child_workflow_id.as_deref(),
            Some(expected),
            "the start is recorded against the parent"
        );
    }

    // The relationship is queryable from the parent's side too.
    let children = reader.get_workflow_children(id).await.expect("read failed");
    assert_eq!(children.len(), 2, "two children: {children:?}");

    dbos.shutdown().await;
}

/// A start dropped part-way still creates the child, records it, and runs it.
///
/// **A start creates a durable object part-way through its own future, and dropping a future does
/// not undo a committed row.** Anything that abandons a branch — a `select!`, a `timeout` — can
/// therefore drop a start after it has written. The writing half runs in a task of its own, so the
/// drop takes the handle and leaves the work, rather than leaving a child with nothing in its
/// parent pointing at it and nothing running it.
///
/// **Polled exactly once and then dropped**, which is what makes this a test rather than a race:
/// one poll is enough to reach the spawn and no further, since everything after it is behind an
/// await. A timeout would prove nothing — it might fire before or after the write.
#[tokio::test]
async fn a_start_dropped_after_it_begins_still_starts_the_child() {
    use std::future::Future;
    use std::pin::Pin;
    use std::task::{Context, Poll, Waker};

    let db = test_database().await;
    let dbos = DBOS::new(config("dropped-start-app", &db));
    let ran = Arc::new(AtomicU32::new(0));
    let child = dbos
        .register_workflow("child", {
            let ran = Arc::clone(&ran);
            move |n: u32| {
                let ran = Arc::clone(&ran);
                async move {
                    ran.fetch_add(1, Ordering::SeqCst);
                    Ok::<u32, Error>(n * 2)
                }
            }
        })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", {
            let child = child.clone();
            move |()| {
                let child = child.clone();
                async move {
                    let mut start = child.start(21);
                    assert_eq!(
                        start.step_id(),
                        Some(0),
                        "the start claimed its id at the call"
                    );
                    // `Pin::new` rather than `pin!`, because a `PendingStep` is `Unpin` and says
                    // so — which is also what lets `start` be dropped outright below rather than
                    // held alive by a pin until the end of this body.
                    // Deterministic, and it is the runtime that makes it so: `#[tokio::test]` is
                    // current-thread, and the only await this poll can reach is the join onto the
                    // task it has just spawned — which cannot have run, because nothing else runs
                    // until this task yields. On a multi-threaded runtime the creation could
                    // finish first and the poll return `Ready`, so a `flavor` here would have to
                    // come with a gate holding the detached task open.
                    assert!(
                        matches!(
                            Pin::new(&mut start).poll(&mut Context::from_waker(Waker::noop())),
                            Poll::Pending
                        ),
                        "one poll should reach the spawn and stop at the await behind it",
                    );
                    drop(start);
                    // The step after it keeps the id it would have had: the dropped start spent
                    // its own, whatever became of the child.
                    dbos::step("after", || async { Ok::<u32, Error>(1) }).await
                }
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let id = "the-dropping-parent";
    parent
        .run_with(
            (),
            RunOptions {
                workflow_id: Some(id),
                ..RunOptions::default()
            },
        )
        .await
        .expect("the parent failed");

    // The parent is done and never awaited the child, so the child settles on its own schedule.
    let reader = reader(&db).await;
    let child_id = format!("{id}-0");
    let deadline = std::time::Instant::now() + Duration::from_secs(30);
    loop {
        if let Some(row) = reader.get_workflow(&child_id).await.expect("read failed")
            && row.status == WorkflowStatus::Success
        {
            assert_eq!(
                row.parent_workflow_id.as_deref(),
                Some(id),
                "the child points back at the parent that dropped its start",
            );
            break;
        }
        assert!(
            std::time::Instant::now() < deadline,
            "the dropped start never produced a finished child",
        );
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    assert_eq!(
        ran.load(Ordering::SeqCst),
        1,
        "the child body ran exactly once"
    );

    // Recorded against the parent as well, so a replay of this position joins the child rather
    // than starting a second one — the half that would be missing if the write had been
    // abandoned rather than detached.
    let steps = reader
        .list_workflow_steps(id, false, None, None, None)
        .await
        .expect("read failed");
    let recorded: Vec<(i32, &str, Option<&str>)> = steps
        .iter()
        .map(|step| {
            (
                step.step_id,
                step.step_name.as_str(),
                step.child_workflow_id.as_deref(),
            )
        })
        .collect();
    assert_eq!(
        recorded,
        [(0, "child", Some(child_id.as_str())), (1, "after", None)],
        "the dropped start recorded its own row and left `after` where it would have been anyway",
    );

    dbos.shutdown().await;
}

/// A start dropped by a parent that then finishes still starts the child, however slow the start.
///
/// **A start writes under its parent's owner token, and a parent recording its outcome gives the
/// token up**, so a start still in flight when its parent finished would be refused and lose the
/// child: nothing awaits the dropped start to hear about it, and nothing records it. The parent
/// therefore waits for its starts before it records its outcome.
///
/// Forced rather than raced: the test holds `operation_outputs` locked, which stops the start at
/// its first read and leaves the parent's outcome write — to `workflow_status` alone — free to go
/// ahead. A parent that recorded its outcome while its start waited would do it here at once.
#[tokio::test]
async fn a_start_dropped_by_a_parent_that_then_finishes_still_starts_the_child() {
    use std::future::Future;
    use std::pin::Pin;
    use std::task::{Context, Poll, Waker};

    let db = test_database().await;
    if db.backend() == Backend::Cockroach {
        // The lock that stalls the start is Postgres's `LOCK TABLE`, and what this checks — the
        // parent waiting for its starts — is the same on both backends.
        return;
    }
    let dbos = DBOS::new(config("outlived-start-app", &db));
    let child = dbos
        .register_workflow("child", |n: u32| async move { Ok::<u32, Error>(n * 2) })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", {
            let child = child.clone();
            move |()| {
                let child = child.clone();
                async move {
                    let mut start = child.start(21);
                    // One poll reaches the spawn and no further; see the test above.
                    assert!(matches!(
                        Pin::new(&mut start).poll(&mut Context::from_waker(Waker::noop())),
                        Poll::Pending
                    ));
                    drop(start);
                    // Nothing after it: a step would write to the locked table and wait too.
                    Ok::<u32, Error>(1)
                }
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let pool = db.pool().await;
    let mut lock = pool.begin().await.expect("begin failed");
    sqlx::query("LOCK TABLE dbos.operation_outputs IN ACCESS EXCLUSIVE MODE")
        .execute(&mut *lock)
        .await
        .expect("lock failed");

    let id = "the-outlived-parent";
    let handle = parent
        .start_with(
            (),
            StartOptions {
                workflow_id: Some(id),
                ..StartOptions::default()
            },
        )
        .await
        .expect("the parent did not start");

    // Long enough for a parent that does not wait to record its outcome many times over.
    let reader = reader(&db).await;
    let deadline = std::time::Instant::now() + Duration::from_secs(2);
    while std::time::Instant::now() < deadline {
        let row = reader.get_workflow(id).await.expect("read failed");
        assert_ne!(
            row.map(|row| row.status),
            Some(WorkflowStatus::Success),
            "the parent recorded its outcome while its start was still in flight",
        );
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    lock.rollback().await.expect("unlock failed");

    assert_eq!(handle.result().await.expect("the parent failed"), 1);
    let child_id = format!("{id}-0");
    let deadline = std::time::Instant::now() + Duration::from_secs(30);
    loop {
        if let Some(row) = reader.get_workflow(&child_id).await.expect("read failed")
            && row.status == WorkflowStatus::Success
        {
            break;
        }
        assert!(
            std::time::Instant::now() < deadline,
            "the start the parent outlived never produced a finished child",
        );
        tokio::time::sleep(Duration::from_millis(50)).await;
    }

    dbos.shutdown().await;
}

/// A parent waiting for a stalled start still times out at its deadline.
///
/// The wait for the parent's starts runs inside its deadline, so a start that cannot finish does
/// not hold the parent open past it: the parent is cancelled as any other would be, and its caller
/// hears so while the start is still stalled. Stalled the same way as the test above.
#[tokio::test]
async fn a_parent_waiting_for_a_stalled_start_still_times_out() {
    use std::future::Future;
    use std::pin::Pin;
    use std::task::{Context, Poll, Waker};

    let db = test_database().await;
    if db.backend() == Backend::Cockroach {
        // `LOCK TABLE` is Postgres's; the deadline it is checked against is the same on both.
        return;
    }
    let dbos = DBOS::new(config("stalled-start-timeout-app", &db));
    let child = dbos
        .register_workflow("child", |n: u32| async move { Ok::<u32, Error>(n * 2) })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", {
            let child = child.clone();
            move |()| {
                let child = child.clone();
                async move {
                    let mut start = child.start(21);
                    assert!(matches!(
                        Pin::new(&mut start).poll(&mut Context::from_waker(Waker::noop())),
                        Poll::Pending
                    ));
                    drop(start);
                    Ok::<u32, Error>(1)
                }
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let pool = db.pool().await;
    let mut lock = pool.begin().await.expect("begin failed");
    sqlx::query("LOCK TABLE dbos.operation_outputs IN ACCESS EXCLUSIVE MODE")
        .execute(&mut *lock)
        .await
        .expect("lock failed");

    let handle = parent
        .start_with(
            (),
            StartOptions {
                workflow_id: Some("the-stalled-parent"),
                timeout: Timeout::Explicit(Duration::from_secs(1)),
                ..StartOptions::default()
            },
        )
        .await
        .expect("the parent did not start");

    // Held until the parent's caller has its answer, so the start is still stalled when it does.
    let result = tokio::time::timeout(Duration::from_secs(20), handle.result())
        .await
        .expect("the parent was held past its deadline by the start it was waiting for");
    assert!(
        matches!(result, Err(Error::WorkflowCancelled { .. })),
        "expected the parent to time out, got {result:?}"
    );
    lock.rollback().await.expect("unlock failed");

    dbos.shutdown().await;
}

/// A child that joins a held deduplication key is recorded as the workflow it joined.
///
/// The interesting half is the start record. The child id this call derived — `{parent}-{step}` —
/// names no row, because the insert lost the key and nothing was written under it. What the parent
/// records at that step is the *holder's* id, so a replay of this position resolves to the same
/// workflow instead of trying to start a child that never existed. Go records the same mapping at
/// the same reserved step id, for the reason it states at `workflow.go`.
#[tokio::test]
async fn a_child_joining_a_held_key_is_recorded_as_the_workflow_it_joined() {
    let db = test_database().await;
    let dbos = DBOS::new(config("child-dedup-app", &db));
    let child = dbos
        .register_workflow("child", |()| async move { Ok::<u32, Error>(9) })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", {
            let child = child.clone();
            move |()| {
                let child = child.clone();
                async move {
                    child
                        .start_with(
                            (),
                            StartOptions {
                                queue: Some(QueueRequest {
                                    deduplication_id: Some("order-42"),
                                    duplication_policy: DuplicationPolicy::ReturnExisting,
                                    ..QueueRequest::new("demo-queue")
                                }),
                                ..StartOptions::default()
                            },
                        )
                        .await?
                        .result()
                        .await
                }
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");
    dbos.register_queue(
        "demo-queue",
        QueueOptions::default(),
        QueueConflict::UpdateIfLatestVersion,
    )
    .await
    .expect("registration failed");

    // The holder, enqueued before the parent runs and still waiting when the child starts.
    let holder = child
        .start_with(
            (),
            StartOptions {
                workflow_id: Some("the-holder"),
                queue: Some(QueueRequest {
                    deduplication_id: Some("order-42"),
                    delay: Some(Duration::from_secs(3)),
                    ..QueueRequest::new("demo-queue")
                }),
                ..StartOptions::default()
            },
        )
        .await
        .expect("the holder was not enqueued");

    let id = "the-joining-parent";
    let result = parent
        .run_with(
            (),
            RunOptions {
                workflow_id: Some(id),
                ..RunOptions::default()
            },
        )
        .await
        .expect("the parent failed");
    assert_eq!(result, 9, "the parent read the joined workflow's output");

    let reader = reader(&db).await;
    assert!(
        reader
            .get_workflow("the-joining-parent-0")
            .await
            .expect("read failed")
            .is_none(),
        "the derived child id names no row: the insert lost the key",
    );
    let steps = reader
        .list_workflow_steps(id, false, None, None, None)
        .await
        .expect("read failed");
    let start = steps
        .iter()
        .find(|step| step.step_id == 0)
        .expect("no start step on the parent");
    assert_eq!(start.step_name, "child");
    assert_eq!(
        start.child_workflow_id.as_deref(),
        Some("the-holder"),
        "the start records the workflow that was joined",
    );

    // The other half of the relationship is deliberately absent. `parent_workflow_id` names the
    // one owner of a row, and the holder has one already -- so a joined workflow is resolved by
    // the parent's replay and named among its steps, but is not listed among its children, and a
    // cascade following that column never reaches it.
    assert!(
        reader
            .get_workflow_children(id)
            .await
            .expect("read failed")
            .is_empty(),
        "a joined workflow is not the joining parent's child",
    );
    assert_eq!(
        reader
            .get_workflow("the-holder")
            .await
            .expect("read failed")
            .expect("the holder is missing")
            .parent_workflow_id,
        None,
        "the holder keeps its own parentage: joining does not re-parent it",
    );

    assert_eq!(
        tokio::time::timeout(Duration::from_secs(20), holder.result())
            .await
            .expect("the holder never ran")
            .expect("the holder failed"),
        9
    );

    dbos.shutdown().await;
}

/// **A deduplicated child start's error is recorded at the parent's step, and replays the same.**
///
/// The key is held only while its holder waits, so a replay after the holder has finished would
/// otherwise start the child and take the branch the first run didn't. The parent handles the
/// error; the holder then finishes; a fork that copies the start's checkpoint still takes the
/// error branch and creates no child. A fork that runs the start again, as a control, starts the
/// child, which shows the key really is free by then.
#[tokio::test]
async fn a_deduplicated_child_start_replays_its_error_after_the_key_is_freed() {
    let db = test_database().await;
    let dbos = DBOS::new(config("child-dedup-error-app", &db));
    let child = dbos
        .register_workflow("child", |()| async move { Ok::<u32, Error>(9) })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", {
            let child = child.clone();
            move |()| {
                let child = child.clone();
                async move {
                    let started = child
                        .start_with(
                            (),
                            StartOptions {
                                queue: Some(QueueRequest {
                                    deduplication_id: Some("order-7"),
                                    ..QueueRequest::new("dedup-error-queue")
                                }),
                                ..StartOptions::default()
                            },
                        )
                        .await;
                    match started {
                        Ok(handle) => Ok::<_, Error>(format!("started {}", handle.result().await?)),
                        Err(
                            e @ Error::SystemDatabase(dbos::sysdb::Error::QueueDeduplicated {
                                ..
                            }),
                        ) => Ok(format!("deduplicated: {e}")),
                        Err(e) => Err(e),
                    }
                }
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    // **Enqueued before the queue is registered, which is what keeps the key held.** Nothing polls
    // a queue until it is registered, so the holder waits, holding the key, for as long as the
    // parent's first run takes.
    let holder = child
        .start_with(
            (),
            StartOptions {
                workflow_id: Some("the-dedup-holder"),
                queue: Some(QueueRequest {
                    deduplication_id: Some("order-7"),
                    ..QueueRequest::new("dedup-error-queue")
                }),
                ..StartOptions::default()
            },
        )
        .await
        .expect("the holder was not enqueued");

    let id = "the-deduplicated-parent";
    let first = parent
        .run_with(
            (),
            RunOptions {
                workflow_id: Some(id),
                ..RunOptions::default()
            },
        )
        .await
        .expect("the parent failed");
    assert!(first.starts_with("deduplicated: "), "{first}");

    let reader = reader(&db).await;
    let steps = reader
        .list_workflow_steps(id, true, None, None, None)
        .await
        .expect("read failed");
    assert_eq!(steps.len(), 1, "{steps:?}");
    assert_eq!(steps[0].step_name, "child");
    assert!(
        steps[0].error.is_some() && steps[0].child_workflow_id.is_none(),
        "the error is recorded as the start's, with no child: {:?}",
        steps[0]
    );

    // Registering the queue lets the holder run, and its finishing frees the key.
    dbos.register_queue(
        "dedup-error-queue",
        QueueOptions::default(),
        QueueConflict::UpdateIfLatestVersion,
    )
    .await
    .expect("registration failed");
    assert_eq!(
        tokio::time::timeout(Duration::from_secs(20), holder.result())
            .await
            .expect("the holder never ran")
            .expect("the holder failed"),
        9
    );

    // The key is free now. A fork from step 1 copies the start's checkpoint, so the start replays
    // and must return the first run's error again rather than start the child: the same variant
    // and fields, so it still names the first run's derived child id.
    let replayed = dbos
        .fork::<String, Error>(id, 1)
        .await
        .expect("fork failed");
    let replayed_id = replayed.workflow_id().to_owned();
    assert_eq!(
        replayed.result().await.expect("the replay failed"),
        first,
        "the replay took a different branch from the run it replayed"
    );
    assert!(
        reader
            .get_workflow(&format!("{replayed_id}-0"))
            .await
            .expect("read failed")
            .is_none(),
        "the replay started the child"
    );

    // A fork from step 0 runs the start again, and with the key free it goes ahead.
    let rerun = dbos
        .fork::<String, Error>(id, 0)
        .await
        .expect("fork failed");
    assert_eq!(
        tokio::time::timeout(Duration::from_secs(20), rerun.result())
            .await
            .expect("the rerun never finished")
            .expect("the rerun failed"),
        "started 9"
    );

    dbos.shutdown().await;
}

/// **A child start the insert refuses for its arguments is recorded too**, and replays the same.
///
/// An empty queue name passes the start's own checks and is refused by the insert, as
/// `InvalidInput`. The name is shared with the test and made valid before the replay, standing in
/// for a refusal the replay would not reach (or a record an earlier version wrote): a fork that
/// copies the start's checkpoint still takes the refused branch and creates no child, and a fork
/// that runs the start again, as a control, starts the child.
#[tokio::test]
async fn a_child_start_refused_by_the_insert_replays_its_error_when_the_start_would_succeed() {
    use std::sync::Mutex;

    let db = test_database().await;
    let dbos = DBOS::new(config("child-insert-error-app", &db));
    let child = dbos
        .register_workflow("child", |()| async move { Ok::<u32, Error>(9) })
        .unwrap();
    let key = Arc::new(Mutex::new(String::new()));
    let parent = dbos
        .register_workflow("parent", {
            let child = child.clone();
            let key = Arc::clone(&key);
            move |()| {
                let child = child.clone();
                let key = key.lock().unwrap().clone();
                async move {
                    let started = child
                        .start_with(
                            (),
                            StartOptions {
                                queue: Some(QueueRequest::new(&key)),
                                ..StartOptions::default()
                            },
                        )
                        .await;
                    match started {
                        Ok(handle) => Ok::<_, Error>(format!("started {}", handle.workflow_id())),
                        Err(e @ Error::InvalidArgument { .. }) => Ok(format!("refused: {e}")),
                        Err(e) => Err(e),
                    }
                }
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let id = "the-insert-refused-parent";
    let first = parent
        .run_with(
            (),
            RunOptions {
                workflow_id: Some(id),
                ..RunOptions::default()
            },
        )
        .await
        .expect("the parent failed");
    assert!(
        first.starts_with("refused: invalid argument to enqueue a workflow: queue ``: `name`"),
        "{first}"
    );

    let reader = reader(&db).await;
    let steps = reader
        .list_workflow_steps(id, true, None, None, None)
        .await
        .expect("read failed");
    assert_eq!(steps.len(), 1, "{steps:?}");
    assert_eq!(steps[0].step_name, "child");
    assert!(
        steps[0].error.is_some() && steps[0].child_workflow_id.is_none(),
        "the refusal is recorded as the start's, with no child: {:?}",
        steps[0]
    );

    *key.lock().unwrap() = "insert-error-queue".to_owned();

    // A fork from step 1 copies the start's checkpoint, so the start replays and must return the
    // first run's refusal again rather than start the child on a queue name that is now valid.
    let replayed = dbos
        .fork::<String, Error>(id, 1)
        .await
        .expect("fork failed");
    let replayed_id = replayed.workflow_id().to_owned();
    // The same refusal, read back from the record. Its message names the queue the caller passes
    // now, which this test has changed, so only the recorded part is compared.
    let replayed = replayed.result().await.expect("the replay failed");
    let refusal = "`name` must be absent rather than empty";
    assert!(
        first.ends_with(refusal)
            && replayed.starts_with("refused: ")
            && replayed.ends_with(refusal),
        "the replay took a different branch from the run it replayed: {replayed}"
    );
    assert!(
        reader
            .get_workflow(&format!("{replayed_id}-0"))
            .await
            .expect("read failed")
            .is_none(),
        "the replay started the child"
    );

    // A fork from step 0 runs the start again, and with a valid queue name it goes ahead.
    let rerun = dbos
        .fork::<String, Error>(id, 0)
        .await
        .expect("fork failed");
    let rerun_id = rerun.workflow_id().to_owned();
    assert_eq!(
        rerun.result().await.expect("the rerun failed"),
        format!("started {rerun_id}-0")
    );

    dbos.shutdown().await;
}

/// **A child id taken by another function is a recorded error too**, and a parent that returns it
/// with `?` ends `ERROR` with it, on the first run and on the replay.
///
/// The conflict lasts only as long as the other workflow's row. Once that row is deleted the same
/// start would go ahead, so a replay that didn't read the record would start the child and
/// end in success.
#[tokio::test]
async fn a_child_id_taken_by_another_function_replays_its_error_after_the_row_is_deleted() {
    let db = test_database().await;
    let dbos = DBOS::new(config("child-conflict-app", &db));
    let child = dbos
        .register_workflow("child", |()| async move { Ok::<u32, Error>(9) })
        .unwrap();
    let squatter = dbos
        .register_workflow("squatter", |()| async move { Ok::<u32, Error>(1) })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", {
            let child = child.clone();
            move |()| {
                let child = child.clone();
                async move {
                    child
                        .start_with(
                            (),
                            StartOptions {
                                workflow_id: Some("the-taken-child-id"),
                                ..StartOptions::default()
                            },
                        )
                        .await?
                        .result()
                        .await
                }
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    squatter
        .run_with(
            (),
            RunOptions {
                workflow_id: Some("the-taken-child-id"),
                ..RunOptions::default()
            },
        )
        .await
        .expect("the squatter failed");

    let id = "the-conflicting-parent";
    let first = parent
        .start_with(
            (),
            StartOptions {
                workflow_id: Some(id),
                ..StartOptions::default()
            },
        )
        .await
        .expect("the parent did not start")
        .result()
        .await
        .expect_err("the parent should end with the error");
    assert!(
        matches!(
            first,
            Error::SystemDatabase(dbos::sysdb::Error::ConflictingWorkflow { .. })
        ),
        "{first:?}"
    );
    let reader = reader(&db).await;
    assert_eq!(
        reader
            .get_workflow(id)
            .await
            .expect("read failed")
            .expect("the parent is missing")
            .status,
        WorkflowStatus::Error,
        "a recorded start error returned with `?` is the workflow's outcome"
    );

    // With the squatter gone the id is free, so only the record keeps the replay failing.
    dbos.delete("the-taken-child-id")
        .await
        .expect("delete failed");
    let fork = dbos.fork::<u32, Error>(id, 1).await.expect("fork failed");
    let fork_id = fork.workflow_id().to_owned();
    let replayed = fork
        .result()
        .await
        .expect_err("the replay should end with the error");
    assert_eq!(
        replayed.to_string(),
        first.to_string(),
        "the replay ended differently from the run it replayed"
    );
    assert_eq!(
        reader
            .get_workflow(&fork_id)
            .await
            .expect("read failed")
            .expect("the fork is missing")
            .status,
        WorkflowStatus::Error,
        "the replay's recorded start error is its outcome too"
    );
    assert!(
        reader
            .get_workflow("the-taken-child-id")
            .await
            .expect("read failed")
            .is_none(),
        "the replay started the child"
    );

    dbos.shutdown().await;
}

/// **A child start that refuses a taken id is a recorded error too**, and a replay ends with the same
/// error after the workflow holding the id is deleted, rather than starting the child.
#[tokio::test]
async fn a_rejected_child_start_replays_its_error_after_the_row_is_deleted() {
    let db = test_database().await;
    let dbos = DBOS::new(config("child-reuse-app", &db));
    let child = dbos
        .register_workflow("child", |()| async move { Ok::<u32, Error>(9) })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", {
            let child = child.clone();
            move |()| {
                let child = child.clone();
                async move {
                    child
                        .start_with(
                            (),
                            StartOptions {
                                workflow_id: Some("the-reused-child-id"),
                                workflow_id_reuse_policy: WorkflowIdReusePolicy::Reject,
                                ..StartOptions::default()
                            },
                        )
                        .await?
                        .result()
                        .await
                }
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    // The same function holds the id: a reuse even so.
    child
        .run_with(
            (),
            RunOptions {
                workflow_id: Some("the-reused-child-id"),
                ..RunOptions::default()
            },
        )
        .await
        .expect("the first child failed");

    let id = "the-rejecting-parent";
    let first = parent
        .start_with(
            (),
            StartOptions {
                workflow_id: Some(id),
                ..StartOptions::default()
            },
        )
        .await
        .expect("the parent did not start")
        .result()
        .await
        .expect_err("the parent should end with the error");
    assert!(
        matches!(
            first,
            Error::SystemDatabase(dbos::sysdb::Error::WorkflowIdInUse { .. })
        ),
        "{first:?}"
    );

    // With the first child gone the id is free, so only the record keeps the replay failing.
    dbos.delete("the-reused-child-id")
        .await
        .expect("delete failed");
    let fork = dbos.fork::<u32, Error>(id, 1).await.expect("fork failed");
    let replayed = fork
        .result()
        .await
        .expect_err("the replay should end with the error");
    assert_eq!(
        replayed.to_string(),
        first.to_string(),
        "the replay ended differently from the run it replayed"
    );
    assert!(
        reader(&db)
            .await
            .get_workflow("the-reused-child-id")
            .await
            .expect("read failed")
            .is_none(),
        "the replay started the child"
    );

    dbos.shutdown().await;
}

/// **The acceptance test.** A parent that dies between its children adopts the ones it already
/// started rather than starting a second set.
///
/// The first process starts two of three children and is killed by shutdown, leaving the parent
/// PENDING. The second process recovers it, and the deterministic id is what makes the starts
/// idempotent: the recovered parent re-derives `parent-0` and `parent-1`, finds them recorded, and
/// runs only the third child's body.
#[tokio::test]
async fn a_recovered_parent_adopts_the_children_it_already_started() {
    let db = test_database().await;
    let id = "crashes-between-children";

    // How many times a child body actually ran, across both processes.
    let bodies = Arc::new(AtomicU32::new(0));

    // First process: two children finish, the third blocks until shutdown kills the parent.
    {
        let dbos = DBOS::new(config("recover-children-app", &db));
        let counted = Arc::clone(&bodies);
        let child = dbos
            .register_workflow("child", move |n: u32| {
                let counted = Arc::clone(&counted);
                async move {
                    counted.fetch_add(1, Ordering::SeqCst);
                    if n == 2 {
                        tokio::time::sleep(Duration::from_secs(30)).await;
                    }
                    Ok::<u32, Error>(n)
                }
            })
            .unwrap();
        let parent = dbos
            .register_workflow("parent", move |()| {
                let child = child.clone();
                async move {
                    let mut total = 0;
                    for n in 0..3 {
                        total += child.run(n).await?;
                    }
                    Ok::<u32, Error>(total)
                }
            })
            .unwrap();
        dbos.launch().await.expect("launch failed");
        parent
            .start_with(
                (),
                StartOptions {
                    workflow_id: Some(id),
                    ..StartOptions::default()
                },
            )
            .await
            .expect("start failed");

        // Wait until the third child's body has entered, so the crash lands mid-parent: two
        // children finished, one in flight.
        //
        // On the counter rather than on the parent's step count, which is the tempting version
        // and is racy. The start row is written *before* the child is spawned — that ordering is
        // what makes a crash in the gap adoptable — so a parent showing five steps can have a
        // third child that has not been polled once, and shutdown would abort it before it
        // counted. Waiting on the thing the next assertion reads leaves no window between them.
        let deadline = std::time::Instant::now() + Duration::from_secs(20);
        while bodies.load(Ordering::SeqCst) < 3 {
            assert!(
                std::time::Instant::now() < deadline,
                "the parent never got as far as its third child"
            );
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
        dbos.shutdown().await;
    }

    let reader = reader(&db).await;
    assert_eq!(
        reader
            .get_workflow(id)
            .await
            .expect("read failed")
            .expect("the row is missing")
            .status,
        WorkflowStatus::Pending,
        "shutdown left the parent for recovery"
    );
    let ran_before = bodies.load(Ordering::SeqCst);
    assert_eq!(
        ran_before, 3,
        "three bodies entered, the third still blocked when the process died"
    );

    // Second process: recovery re-runs the parent, which must adopt rather than start again.
    let dbos = DBOS::new(config("recover-children-app", &db));
    let recovered_bodies = Arc::clone(&bodies);
    let child = dbos
        .register_workflow("child", move |n: u32| {
            let bodies = Arc::clone(&recovered_bodies);
            async move {
                bodies.fetch_add(1, Ordering::SeqCst);
                Ok::<u32, Error>(n)
            }
        })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", move |()| {
            let child = child.clone();
            async move {
                let mut total = 0;
                for n in 0..3 {
                    total += child.run(n).await?;
                }
                Ok::<u32, Error>(total)
            }
        })
        .unwrap();
    // Registered so recovery can find it; nothing here starts it, recovery does.
    dbos.launch().await.expect("launch failed");
    drop(parent);

    // The parent finishes, and its children are the same three rows.
    let deadline = std::time::Instant::now() + Duration::from_secs(30);
    loop {
        let row = reader
            .get_workflow(id)
            .await
            .expect("read failed")
            .expect("the row is missing");
        if row.status == WorkflowStatus::Success {
            break;
        }
        assert!(
            std::time::Instant::now() < deadline,
            "the recovered parent did not finish; status {:?}",
            row.status
        );
        tokio::time::sleep(Duration::from_millis(100)).await;
    }

    let children = reader.get_workflow_children(id).await.expect("read failed");
    assert_eq!(
        children.len(),
        3,
        "the recovered parent started no fourth child: {children:?}"
    );
    let steps = reader
        .list_workflow_steps(id, false, None, None, None)
        .await
        .expect("read failed");
    assert_eq!(
        steps.len(),
        6,
        "a start and an await per child, and no second set: {steps:?}"
    );
    assert_eq!(
        bodies.load(Ordering::SeqCst),
        ran_before + 1,
        "only the unfinished child ran again: the two that succeeded were adopted, not re-run"
    );

    dbos.shutdown().await;
}

/// The fan-out shape: children are started one at a time and awaited one at a time, but they
/// **run** concurrently.
///
/// Sequential bookkeeping is the step counter's determinism constraint reaching a second caller —
/// each start and each await allocates a step id, and concurrent allocation would replay against
/// the wrong slots. It costs nothing in wall-clock: three children that each sleep are all in
/// flight together, so the parent takes about as long as the slowest rather than the sum.
///
/// A `join!` over the child starts — or over the awaits, or over whole `run`s — is sound, because
/// a start takes its id where it is *built*, the same as a step: `join!` builds every branch
/// before polling any, which is exactly the order the ids were claimed in. The loop is here
/// because it is the plainest way to write a fan-out, not because it is the only sound one; the
/// tests below drive the concurrent forms out of build order on purpose.
#[tokio::test]
async fn children_started_in_a_loop_run_concurrently() {
    let db = test_database().await;
    let dbos = DBOS::new(config("fan-out-app", &db));
    let child = dbos
        .register_workflow("child", |n: u32| async move {
            tokio::time::sleep(Duration::from_millis(400)).await;
            Ok::<u32, Error>(n)
        })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", move |()| {
            let child = child.clone();
            async move {
                // Start all three first — each `start` returns as soon as the child is recorded
                // and spawned — then collect. Awaiting inside the first loop would serialize them.
                let mut handles = Vec::new();
                for n in 0..3 {
                    handles.push(child.start(n).await?);
                }
                let mut total = 0;
                for handle in handles {
                    total += handle.result().await?;
                }
                Ok::<u32, Error>(total)
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let began = std::time::Instant::now();
    let total = parent
        .run_with(
            (),
            RunOptions {
                workflow_id: Some("fans-out"),
                ..RunOptions::default()
            },
        )
        .await
        .expect("the parent failed");
    let took = began.elapsed();

    assert_eq!(total, 3, "0 + 1 + 2");
    assert!(
        took < Duration::from_millis(1_200),
        "three 400ms children took {took:?}, which is serial rather than concurrent"
    );

    let children = reader(&db)
        .await
        .get_workflow_children("fans-out")
        .await
        .expect("read failed");
    assert_eq!(children.len(), 3);

    dbos.shutdown().await;
}

/// Starts and awaits driven out of build order keep the ids they were built with.
///
/// The test that says the ids are taken at the **call** rather than at the first poll, and the
/// only shape that says it. Branches that differ only by how long their children sleep prove
/// nothing: a sleep changes which child *finishes* first, which no id depends on, and `join!`
/// first-polls its branches in source order — so a start that took its id at the first poll would
/// pass such a test too. Here the three starts are built `a, b, c` and handed to `join!` as
/// `c, b, a`, and then the three awaits the same way. Under poll-time ids `c` would take step 0
/// and the child ids would come out reversed.
///
/// Both halves are asserted from the rows rather than from inside the body, because a failed
/// assertion inside a workflow is a panicking task rather than a failing test.
#[tokio::test]
async fn child_starts_and_awaits_driven_out_of_build_order_keep_their_ids() {
    let db = test_database().await;
    let dbos = DBOS::new(config("out-of-order-children-app", &db));
    let child = dbos
        .register_workflow("child", |n: u32| async move { Ok::<u32, Error>(n) })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", move |()| {
            let child = child.clone();
            async move {
                // Built `a, b, c`, so `a` holds step 0 whatever happens next.
                let a = child.start(1);
                let b = child.start(2);
                let c = child.start(3);
                // ...and driven the other way round.
                let (c, b, a) = tokio::join!(c, b, a);
                let (a, b, c) = (a?, b?, c?);
                // The awaits are their own three ids, built and driven the same way.
                let a = a.result();
                let b = b.result();
                let c = c.result();
                let (c, b, a) = tokio::join!(c, b, a);
                Ok::<u32, Error>(a? + b? + c?)
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let id = "drives-its-children-backwards";
    let total = parent
        .run_with(
            (),
            RunOptions {
                workflow_id: Some(id),
                ..RunOptions::default()
            },
        )
        .await
        .expect("the parent failed");
    assert_eq!(total, 6, "1 + 2 + 3");

    let reader = reader(&db).await;
    let steps = reader
        .list_workflow_steps(id, false, None, None, None)
        .await
        .expect("read failed");
    let recorded: Vec<(i32, &str, Option<&str>)> = steps
        .iter()
        .map(|step| {
            (
                step.step_id,
                step.step_name.as_str(),
                step.child_workflow_id.as_deref(),
            )
        })
        .collect();
    assert_eq!(
        recorded,
        [
            (0, "child", Some("drives-its-children-backwards-0")),
            (1, "child", Some("drives-its-children-backwards-1")),
            (2, "child", Some("drives-its-children-backwards-2")),
            (3, "DBOS.getResult", Some("drives-its-children-backwards-0")),
            (4, "DBOS.getResult", Some("drives-its-children-backwards-1")),
            (5, "DBOS.getResult", Some("drives-its-children-backwards-2")),
        ],
        "the three starts hold the first three ids and the three awaits the next three, both in \
         the order they were written"
    );

    // Which child is which: the first-built start passed 1, so `{parent}-0` is the child that
    // returned 1. This is what a reversed numbering would show up as.
    for (suffix, expected) in [("-0", "1"), ("-1", "2"), ("-2", "3")] {
        let child_id = format!("{id}{suffix}");
        let row = reader
            .get_workflow(&child_id)
            .await
            .expect("read failed")
            .unwrap_or_else(|| panic!("no row for {child_id}"));
        assert_eq!(
            row.output.as_deref(),
            Some(expected),
            "{child_id} ran the argument the start built at that position"
        );
    }

    dbos.shutdown().await;
}

/// Runs driven out of build order keep their `{start, await}` pairs.
///
/// A `run` claims two ids, and it claims them together: the await's immediately behind the
/// start's. So three runs written `a, b, c` occupy `(0, 1)`, `(2, 3)`, `(4, 5)` however they are
/// then driven — and the same three under an await placed when the child's future resolved would
/// interleave the pairs by whichever child answered first. `join!` is handed them backwards for
/// the same reason the test above does.
#[tokio::test]
async fn runs_driven_out_of_build_order_keep_their_pairs_of_step_ids() {
    let db = test_database().await;
    let dbos = DBOS::new(config("out-of-order-runs-app", &db));
    let child = dbos
        .register_workflow("child", |n: u32| async move { Ok::<u32, Error>(n) })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", move |()| {
            let child = child.clone();
            async move {
                let a = child.run(1);
                let b = child.run(2);
                let c = child.run(3);
                let (c, b, a) = tokio::join!(c, b, a);
                Ok::<u32, Error>(a? + b? + c?)
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let id = "runs-its-children-backwards";
    let total = parent
        .run_with(
            (),
            RunOptions {
                workflow_id: Some(id),
                ..RunOptions::default()
            },
        )
        .await
        .expect("the parent failed");
    assert_eq!(total, 6, "1 + 2 + 3");

    let steps = reader(&db)
        .await
        .list_workflow_steps(id, false, None, None, None)
        .await
        .expect("read failed");
    let recorded: Vec<(i32, &str, Option<&str>)> = steps
        .iter()
        .map(|step| {
            (
                step.step_id,
                step.step_name.as_str(),
                step.child_workflow_id.as_deref(),
            )
        })
        .collect();
    assert_eq!(
        recorded,
        [
            (0, "child", Some("runs-its-children-backwards-0")),
            (1, "DBOS.getResult", Some("runs-its-children-backwards-0")),
            (2, "child", Some("runs-its-children-backwards-2")),
            (3, "DBOS.getResult", Some("runs-its-children-backwards-2")),
            (4, "child", Some("runs-its-children-backwards-4")),
            (5, "DBOS.getResult", Some("runs-its-children-backwards-4")),
        ],
        "each run's await sits immediately behind its own start"
    );

    dbos.shutdown().await;
}

/// A child started and never awaited is still recorded, so a recovered parent adopts it.
///
/// The start row is what makes a child adoptable, and it is written by `start` alone — nothing
/// about awaiting the handle is what records the relationship.
#[tokio::test]
async fn a_child_that_is_never_awaited_is_still_recorded() {
    let db = test_database().await;
    let dbos = DBOS::new(config("unawaited-child-app", &db));
    let child = dbos
        .register_workflow("child", |()| async move { Ok::<u32, Error>(1) })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", move |()| {
            let child = child.clone();
            async move {
                // Started, and the handle dropped without ever being awaited.
                child.start(()).await?;
                Ok::<u32, Error>(0)
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let id = "forgets-its-child";
    parent
        .run_with(
            (),
            RunOptions {
                workflow_id: Some(id),
                ..RunOptions::default()
            },
        )
        .await
        .expect("the parent failed");

    let reader = reader(&db).await;
    let steps = reader
        .list_workflow_steps(id, false, None, None, None)
        .await
        .expect("read failed");
    assert_eq!(steps.len(), 1, "the start is recorded: {steps:?}");
    assert_eq!(
        steps[0].child_workflow_id.as_deref(),
        Some("forgets-its-child-0")
    );

    // The child outlives the parent's interest in it and finishes on its own.
    let deadline = std::time::Instant::now() + Duration::from_secs(10);
    loop {
        let row = reader
            .get_workflow("forgets-its-child-0")
            .await
            .expect("read failed")
            .expect("the child has no row");
        if row.status == WorkflowStatus::Success {
            break;
        }
        assert!(
            std::time::Instant::now() < deadline,
            "the abandoned child never finished; status {:?}",
            row.status
        );
        tokio::time::sleep(Duration::from_millis(50)).await;
    }

    dbos.shutdown().await;
}

/// A child cannot be started from inside a step.
///
/// A step is a leaf, and an id-allocating call inside one shifts every later step onto the wrong
/// replay slot. There is no plain version of a durable start to degrade to, so this is an error
/// rather than the quiet fallback a nested *step* gets.
#[tokio::test]
async fn a_child_cannot_be_started_from_inside_a_step() {
    let db = test_database().await;
    let dbos = DBOS::new(config("child-in-step-app", &db));
    let child = dbos
        .register_workflow("child", |()| async move { Ok::<u32, Error>(1) })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", move |()| {
            let child = child.clone();
            async move {
                dbos::step::<u32, dbos::EngineOnly, _, _>("tries_to_spawn", || {
                    let child = child.clone();
                    async move { child.run(()).await }
                })
                .await
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let error = parent
        .run_with(
            (),
            RunOptions {
                workflow_id: Some("spawns-in-a-step"),
                ..RunOptions::default()
            },
        )
        .await
        .expect_err("the parent succeeded");
    assert!(
        matches!(&error, Error::InsideStep { operation } if operation.contains("workflow")),
        "expected an inside-a-step refusal, got {error:?}"
    );

    dbos.shutdown().await;
}

/// An application-assigned id wins over the derivation, and the start is still recorded.
#[tokio::test]
async fn an_assigned_child_id_wins_over_the_derived_one() {
    let db = test_database().await;
    let dbos = DBOS::new(config("child-assigned-id-app", &db));
    let child = dbos
        .register_workflow("child", |()| async move { Ok::<u32, Error>(7) })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", move |()| {
            let child = child.clone();
            async move {
                child
                    .run_with(
                        (),
                        RunOptions {
                            workflow_id: Some("i-named-this-one"),
                            ..RunOptions::default()
                        },
                    )
                    .await
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let id = "names-its-child";
    assert_eq!(
        parent
            .run_with(
                (),
                RunOptions {
                    workflow_id: Some(id),
                    ..RunOptions::default()
                },
            )
            .await
            .expect("the parent failed"),
        7
    );

    let reader = reader(&db).await;
    let row = reader
        .get_workflow("i-named-this-one")
        .await
        .expect("read failed")
        .expect("the assigned id has no row");
    assert_eq!(row.parent_workflow_id.as_deref(), Some(id));
    assert!(
        reader
            .get_workflow("names-its-child-0")
            .await
            .expect("read failed")
            .is_none(),
        "the derived id was not used as well"
    );

    dbos.shutdown().await;
}

/// A workflow started outside any workflow is a root: no parent, and no start recorded anywhere.
#[tokio::test]
async fn a_workflow_started_outside_a_workflow_has_no_parent() {
    let db = test_database().await;
    let dbos = DBOS::new(config("root-app", &db));
    let root = dbos
        .register_workflow("root", |()| async move { Ok::<u32, Error>(1) })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let id = "a-root";
    root.run_with(
        (),
        RunOptions {
            workflow_id: Some(id),
            ..RunOptions::default()
        },
    )
    .await
    .expect("the workflow failed");

    let row = reader(&db)
        .await
        .get_workflow(id)
        .await
        .expect("read failed")
        .expect("the row is missing");
    assert!(row.parent_workflow_id.is_none());

    dbos.shutdown().await;
}

/// Awaiting a child is itself a step: the parent records what the child returned.
///
/// It is recorded as `DBOS.getResult`. The row carries the child's own encoded output, not a
/// re-encoding of it, and the child id beside it.
#[tokio::test]
async fn awaiting_a_child_is_recorded_as_a_step() {
    let db = test_database().await;
    let dbos = DBOS::new(config("await-checkpoint-app", &db));
    let child = dbos
        .register_workflow("child", |()| async move { Ok::<u32, Error>(99) })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", move |()| {
            let child = child.clone();
            async move { child.run(()).await }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let id = "records-its-await";
    assert_eq!(
        parent
            .run_with(
                (),
                RunOptions {
                    workflow_id: Some(id),
                    ..RunOptions::default()
                },
            )
            .await
            .expect("the parent failed"),
        99
    );

    let steps = reader(&db)
        .await
        .list_workflow_steps(id, true, None, None, None)
        .await
        .expect("read failed");
    assert_eq!(steps.len(), 2, "one start, one await: {steps:?}");
    assert_eq!(steps[0].step_name, "child", "the start");
    assert_eq!(steps[1].step_name, "DBOS.getResult", "the await");
    assert_eq!(steps[1].output.as_deref(), Some("99"));
    assert_eq!(
        steps[1].child_workflow_id.as_deref(),
        Some("records-its-await-0"),
        "the await says which child it was waiting on"
    );

    dbos.shutdown().await;
}

/// A replayed parent takes the recorded outcome instead of waiting on the child again.
///
/// The child is deleted out from under the recovered parent, which is how the test tells the two
/// apart: a workflow that no longer exists cannot be awaited, so finishing anyway is only possible
/// from the recorded value. **The deletion is the instrument, not the subject.** Removing a
/// workflow another one is still replaying through is outside what the engine defines — this pins
/// where the value is read from, not a promise that a parent survives losing its children.
#[tokio::test]
async fn a_replayed_parent_reads_the_recorded_outcome_rather_than_waiting_again() {
    let db = test_database().await;
    let id = "already-knows";
    let reader = reader(&db).await;

    // First process: the child finishes and its result is recorded, then the parent is killed
    // before it can finish.
    {
        let dbos = DBOS::new(config("replay-await-app", &db));
        let child = dbos
            .register_workflow("child", |()| async move { Ok::<u32, Error>(5) })
            .unwrap();
        let parent = dbos
            .register_workflow("parent", move |()| {
                let child = child.clone();
                async move {
                    let value = child.run(()).await?;
                    // Long enough that shutdown lands after the await is recorded.
                    tokio::time::sleep(Duration::from_secs(30)).await;
                    Ok::<u32, Error>(value)
                }
            })
            .unwrap();
        dbos.launch().await.expect("launch failed");
        parent
            .start_with(
                (),
                StartOptions {
                    workflow_id: Some(id),
                    ..StartOptions::default()
                },
            )
            .await
            .expect("start failed");

        let deadline = std::time::Instant::now() + Duration::from_secs(20);
        loop {
            let steps = reader
                .list_workflow_steps(id, false, None, None, None)
                .await
                .expect("read failed");
            if steps.len() == 2 {
                break;
            }
            assert!(
                std::time::Instant::now() < deadline,
                "the await was never recorded: {steps:?}"
            );
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
        dbos.shutdown().await;
    }

    // The child is gone. Only the parent's recorded copy of its result is left.
    reader
        .delete_workflows(&[&format!("{id}-0")], false, None)
        .await
        .expect("delete failed");
    assert!(
        reader
            .get_workflow(&format!("{id}-0"))
            .await
            .expect("read failed")
            .is_none(),
        "the child was deleted"
    );

    // Second process: recovery replays the parent, which must not go looking for the child.
    let dbos = DBOS::new(config("replay-await-app", &db));
    let child = dbos
        .register_workflow("child", |()| async move { Ok::<u32, Error>(5) })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", move |()| {
            let child = child.clone();
            async move { child.run(()).await }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");
    drop(parent);

    let deadline = std::time::Instant::now() + Duration::from_secs(20);
    loop {
        let row = reader
            .get_workflow(id)
            .await
            .expect("read failed")
            .expect("the row is missing");
        if row.status == WorkflowStatus::Success {
            assert_eq!(row.output.as_deref(), Some("5"), "from the recorded await");
            break;
        }
        assert!(
            std::time::Instant::now() < deadline,
            "the replayed parent did not finish; status {:?}",
            row.status
        );
        tokio::time::sleep(Duration::from_millis(100)).await;
    }

    dbos.shutdown().await;
}

/// A cancelled child is reported to its parent as an *awaited* cancellation, not as the parent
/// being cancelled — and that outcome is recorded like any other.
#[tokio::test]
async fn a_cancelled_child_is_an_awaited_cancellation_in_the_parent() {
    let db = test_database().await;
    let dbos = DBOS::new(config("cancelled-child-app", &db));
    let child = dbos
        .register_workflow("child", |()| async move {
            tokio::time::sleep(Duration::from_secs(30)).await;
            Ok::<u32, Error>(1)
        })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", move |()| {
            let child = child.clone();
            async move {
                // Its own budget, so the child cancels itself while the parent waits.
                child
                    .run_with(
                        (),
                        RunOptions {
                            timeout: Timeout::Explicit(Duration::from_millis(300)),
                            ..RunOptions::default()
                        },
                    )
                    .await
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let id = "waits-on-a-doomed-child";
    let error = parent
        .run_with(
            (),
            RunOptions {
                workflow_id: Some(id),
                ..RunOptions::default()
            },
        )
        .await
        .expect_err("the parent succeeded");
    match &error {
        Error::AwaitedWorkflowCancelled { workflow_id } => {
            assert_eq!(
                workflow_id, "waits-on-a-doomed-child-0",
                "the child, not the parent"
            );
        }
        other => panic!("expected an awaited cancellation, got {other:?}"),
    }

    let reader = reader(&db).await;
    let steps = reader
        .list_workflow_steps(id, true, None, None, None)
        .await
        .expect("read failed");
    let await_step = steps
        .iter()
        .find(|step| step.step_name == "DBOS.getResult")
        .expect("the await was not recorded");
    let recorded = await_step.error.as_deref().expect("no error recorded");
    assert!(
        recorded.contains("AwaitedWorkflowCancelled"),
        "the row says the awaited workflow was cancelled: {recorded}"
    );

    // The parent failed with it rather than being cancelled itself.
    let row = reader
        .get_workflow(id)
        .await
        .expect("read failed")
        .expect("the row is missing");
    assert_eq!(
        row.status,
        WorkflowStatus::Error,
        "the parent failed; it was not itself cancelled"
    );

    dbos.shutdown().await;
}

/// A child with no budget of its own inherits its parent's deadline, as the same instant.
///
/// Not "the same duration again": the child's row carries the parent's deadline verbatim, so a
/// parent an hour into a two-hour budget hands its child one hour, not two.
#[tokio::test]
async fn a_child_inherits_its_parents_deadline() {
    let db = test_database().await;
    let dbos = DBOS::new(config("inherit-deadline-app", &db));
    let child = dbos
        .register_workflow("child", |()| async move { Ok::<u32, Error>(1) })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", move |()| {
            let child = child.clone();
            async move { child.run(()).await }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let id = "hands-down-its-deadline";
    parent
        .run_with(
            (),
            RunOptions {
                workflow_id: Some(id),
                timeout: Timeout::Explicit(Duration::from_secs(300)),
                ..Default::default()
            },
        )
        .await
        .expect("the parent failed");

    let reader = reader(&db).await;
    let parent_row = reader
        .get_workflow(id)
        .await
        .expect("read failed")
        .expect("the row is missing");
    let child_row = reader
        .get_workflow(&format!("{id}-0"))
        .await
        .expect("read failed")
        .expect("the child has no row");
    assert!(parent_row.deadline.is_some(), "the parent has a deadline");
    assert_eq!(
        child_row.deadline, parent_row.deadline,
        "the same instant, not a fresh budget"
    );
    assert!(
        child_row.timeout.is_none(),
        "the child was given no timeout of its own; it has a deadline because its parent had one"
    );

    dbos.shutdown().await;
}

/// A child's own timeout replaces the inherited deadline, even when it is the longer of the two.
///
/// An explicitly specified timeout is used over any propagated deadline. The visible consequence is
/// that such a child **outlives its parent** — which is the point of asking for a timeout on one
/// specific child.
#[tokio::test]
async fn a_childs_own_timeout_replaces_the_inherited_deadline() {
    let db = test_database().await;
    let dbos = DBOS::new(config("child-timeout-app", &db));
    let child = dbos
        .register_workflow("child", |()| async move { Ok::<u32, Error>(1) })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", move |()| {
            let child = child.clone();
            async move {
                child
                    .run_with(
                        (),
                        RunOptions {
                            // Longer than what the parent has left.
                            timeout: Timeout::Explicit(Duration::from_secs(3_600)),
                            ..RunOptions::default()
                        },
                    )
                    .await
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let id = "child-asks-for-more";
    parent
        .run_with(
            (),
            RunOptions {
                workflow_id: Some(id),
                timeout: Timeout::Explicit(Duration::from_secs(60)),
                ..Default::default()
            },
        )
        .await
        .expect("the parent failed");

    let reader = reader(&db).await;
    let parent_deadline = reader
        .get_workflow(id)
        .await
        .expect("read failed")
        .expect("the row is missing")
        .deadline
        .expect("the parent has a deadline");
    let child_row = reader
        .get_workflow(&format!("{id}-0"))
        .await
        .expect("read failed")
        .expect("the child has no row");
    assert!(
        child_row.deadline.expect("the child has a deadline") > parent_deadline,
        "the child's own timeout won: it outlives its parent"
    );
    assert_eq!(child_row.timeout, Some(Duration::from_secs(3_600)));

    dbos.shutdown().await;
}

/// A child can decline the inherited deadline outright, which no `Option<Duration>` could say.
///
/// [`Timeout::Inherit`] is a caller who said nothing and [`Timeout::None`] is a caller who decided,
/// and only the second detaches a child from a parent that has a budget: it clears the propagated
/// deadline rather than merely leaving it unset.
///
/// The row is the assertion: no deadline, and no timeout either. A `workflow_timeout_ms` here
/// would be a budget nobody asked for, and it is the column a queue recomputes a deadline from.
#[tokio::test]
async fn a_child_can_decline_the_inherited_deadline() {
    let db = test_database().await;
    let dbos = DBOS::new(config("child-no-timeout-app", &db));
    let child = dbos
        .register_workflow("child", |()| async move { Ok::<u32, Error>(1) })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", move |()| {
            let child = child.clone();
            async move {
                // Two children under one bounded parent: the first says nothing, the second
                // declines. Together they are the difference the enum exists for.
                let inherited = child.run(()).await?;
                let detached = child
                    .run_with(
                        (),
                        RunOptions {
                            timeout: Timeout::None,
                            ..RunOptions::default()
                        },
                    )
                    .await?;
                Ok::<u32, Error>(inherited + detached)
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let id = "one-child-opts-out";
    let total = parent
        .run_with(
            (),
            RunOptions {
                workflow_id: Some(id),
                timeout: Timeout::Explicit(Duration::from_secs(300)),
                ..Default::default()
            },
        )
        .await
        .expect("the parent failed");
    assert_eq!(total, 2, "both children ran");

    let reader = reader(&db).await;
    let parent_deadline = reader
        .get_workflow(id)
        .await
        .expect("read failed")
        .expect("the row is missing")
        .deadline
        .expect("the parent has a deadline");

    let inherited = reader
        .get_workflow(&format!("{id}-0"))
        .await
        .expect("read failed")
        .expect("the first child has no row");
    assert_eq!(
        inherited.deadline,
        Some(parent_deadline),
        "silence inherits the parent's instant verbatim"
    );

    let detached = reader
        .get_workflow(&format!("{id}-2"))
        .await
        .expect("read failed")
        .expect("the second child has no row");
    assert_eq!(
        detached.deadline, None,
        "Timeout::None declined the parent's deadline"
    );
    assert_eq!(
        detached.timeout, None,
        "and took no budget of its own in its place"
    );

    dbos.shutdown().await;
}

/// **The propagated deadline is the cascade.** A parent that runs out of time and the child that
/// inherited its deadline are cancelled at the same instant, independently.
///
/// Nothing signals the child. It holds the same instant, its own timer fires on it, and it records
/// its own timeout. That is why deadlines need no cancellation cascade: there is
/// nothing left for one to do. (An explicit `cancel` with children is a different question, and
/// belongs to the management surface.)
#[tokio::test]
async fn a_parent_and_its_child_hit_an_inherited_deadline_independently() {
    let db = test_database().await;
    let dbos = DBOS::new(config("deadline-cascade-app", &db));
    let child = dbos
        .register_workflow("child", |()| async move {
            tokio::time::sleep(Duration::from_secs(30)).await;
            Ok::<u32, Error>(1)
        })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", move |()| {
            let child = child.clone();
            async move { child.run(()).await }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let id = "runs-out-of-time";
    let error = parent
        .run_with(
            (),
            RunOptions {
                workflow_id: Some(id),
                timeout: Timeout::Explicit(Duration::from_millis(400)),
                ..Default::default()
            },
        )
        .await
        .expect_err("the parent succeeded");
    assert!(
        matches!(error, Error::WorkflowCancelled { .. }),
        "the parent was cancelled by its own deadline, got {error:?}"
    );

    // The child cancels itself on the same instant, with nothing telling it to.
    let reader = reader(&db).await;
    let deadline = std::time::Instant::now() + Duration::from_secs(10);
    loop {
        let row = reader
            .get_workflow(&format!("{id}-0"))
            .await
            .expect("read failed")
            .expect("the child has no row");
        if row.status == WorkflowStatus::Cancelled {
            break;
        }
        assert!(
            std::time::Instant::now() < deadline,
            "the child was not cancelled by the deadline it inherited; status {:?}",
            row.status
        );
        tokio::time::sleep(Duration::from_millis(50)).await;
    }

    // And the parent, cancelled *while awaiting*, checkpointed nothing for that await. The wait
    // was interrupted rather than answered, so a resumed parent asks the child's then-settled row
    // again instead of replaying a verdict it never actually received.
    let steps = reader
        .list_workflow_steps(id, false, None, None, None)
        .await
        .expect("read failed");
    assert_eq!(
        steps.len(),
        1,
        "the start, and nothing for the interrupted await: {steps:?}"
    );
    assert_eq!(steps[0].step_name, "child", "the start");

    dbos.shutdown().await;
}

/// A start position that already holds a *plain* step is an error, not a second child.
///
/// **Refused before anything is created, deliberately.** Falling through to a fresh child would
/// collide on the step write a moment later, so it would end up loud rather than wrong — but only
/// after creating and orphaning a child workflow. Refusing first leaves nothing behind to clean up.
///
/// The situation is a parent whose code changed under it: `step("child")` at position 0 became
/// `child.run(())`. Planted directly here rather than staged across two processes, because the
/// database is the whole of what the replay reads and the gate is what makes the write land in
/// the same place a crash would have left it.
#[tokio::test]
async fn a_start_position_holding_a_plain_step_is_refused() {
    let db = test_database().await;
    let dbos = DBOS::new(config("stale-start-app", &db));
    let child = dbos
        .register_workflow("child", |()| async move { Ok::<u32, Error>(1) })
        .unwrap();

    // The parent waits before its first step id is allocated, which is the window the plain step
    // is planted in. Nothing before this is a step, so the start still lands on position 0.
    let (tx, rx) = tokio::sync::oneshot::channel::<()>();
    let gate = Arc::new(std::sync::Mutex::new(Some(rx)));
    let parent = dbos
        .register_workflow("parent", move |()| {
            let child = child.clone();
            let gate = Arc::clone(&gate);
            async move {
                let rx = gate.lock().unwrap().take().expect("the parent runs once");
                let _ = rx.await;
                child.run(()).await
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let id = "changed-under-itself";
    let handle = parent
        .start_with(
            (),
            StartOptions {
                workflow_id: Some(id),
                ..StartOptions::default()
            },
        )
        .await
        .expect("start failed");

    // The row exists as soon as `start_with` returns, which is what this write needs.
    let reader = reader(&db).await;
    reader
        .record_step(
            id,
            0,
            "child",
            Outcome::Output(Some("1")),
            None,
            None,
            &owner_token(&reader, id).await,
        )
        .await
        .expect("planting the step failed");
    tx.send(()).expect("the parent is waiting on the gate");

    let error = handle.result().await.expect_err("the parent succeeded");
    match &error {
        Error::SystemDatabase(dbos::sysdb::Error::UnexpectedStep {
            step_id,
            expected,
            recorded,
            ..
        }) => {
            assert_eq!(*step_id, 0, "the start position");
            assert!(
                expected.contains("child workflow start"),
                "says what it wanted: {expected}"
            );
            assert!(
                recorded.contains("plain step"),
                "and what it found: {recorded}"
            );
        }
        other => panic!("expected an unexpected-step refusal, got {other:?}"),
    }

    // The point of refusing early: nothing was created to be orphaned.
    assert!(
        reader
            .get_workflow(&format!("{id}-0"))
            .await
            .expect("read failed")
            .is_none(),
        "no child row was created before the refusal"
    );

    dbos.shutdown().await;
}

/// A child start whose timeout or delay the row cannot hold is refused before it takes a step id.
///
/// The same goes for a zero timeout or delay, which no row can carry either. All of them are
/// refused with the start's other argument checks, ahead of the placement, so the
/// parent's counter does not move: the step written after them takes the id the first start would
/// have had, and no child row is created under any derived id.
#[tokio::test]
async fn a_child_start_refused_for_an_unstorable_duration_spends_no_step_id() {
    let db = test_database().await;
    let dbos = DBOS::new(config("refused-duration-child-app", &db));
    let child = dbos
        .register_workflow("child", |()| async move { Ok::<u32, Error>(1) })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", move |()| {
            let child = child.clone();
            async move {
                // Too long to store, and zero: neither is a timeout or delay a row can carry.
                for duration in [Duration::MAX, Duration::ZERO] {
                    let refused = child
                        .start_with(
                            (),
                            StartOptions {
                                timeout: Timeout::Explicit(duration),
                                ..StartOptions::default()
                            },
                        )
                        .await;
                    assert!(
                        matches!(&refused, Err(Error::InvalidArgument { operation, detail })
                            if operation == "start a workflow" && detail.contains("`timeout`")),
                        "a timeout of {duration:?} should be refused, got {:?}",
                        refused.map(|_| ())
                    );

                    let refused = child
                        .start_with(
                            (),
                            StartOptions {
                                queue: Some(QueueRequest {
                                    delay: Some(duration),
                                    ..QueueRequest::new("never-reached")
                                }),
                                ..StartOptions::default()
                            },
                        )
                        .await;
                    assert!(
                        matches!(&refused, Err(Error::InvalidArgument { operation, detail })
                            if operation == "enqueue a workflow" && detail.contains("`delay`")),
                        "a delay of {duration:?} should be refused, got {:?}",
                        refused.map(|_| ())
                    );
                }

                dbos::step("after", || async { Ok(()) }).await?;
                Ok::<u32, Error>(1)
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let id = "refuses-its-children";
    parent
        .run_with(
            (),
            RunOptions {
                workflow_id: Some(id),
                ..RunOptions::default()
            },
        )
        .await
        .expect("the parent failed");

    let reader = reader(&db).await;
    let steps = reader
        .list_workflow_steps(id, true, None, None, None)
        .await
        .expect("read failed");
    let recorded: Vec<(i32, &str)> = steps
        .iter()
        .map(|s| (s.step_id, s.step_name.as_str()))
        .collect();
    assert_eq!(
        recorded,
        [(0, "after")],
        "a refused start took the step id the step after it should have had",
    );
    for child_id in (0..4).map(|step| format!("{id}-{step}")) {
        assert!(
            reader
                .get_workflow(&child_id)
                .await
                .expect("read failed")
                .is_none(),
            "a refused start created {child_id}"
        );
    }

    dbos.shutdown().await;
}

/// A `WorkflowRef` from another instance cannot start a child of this workflow.
///
/// The second place [`Error::WrongInstance`] is reachable from, and the reason it exists: the step
/// id would come from this workflow's counter while the start record went through the other
/// instance's system database, landing where the workflow that allocated it cannot see it.
/// `DBOS::get_event` refuses the same combination for the same reason.
#[tokio::test]
async fn a_child_started_through_another_instance_is_refused() {
    let db = test_database().await;
    let other = DBOS::new(config("other-instance-app", &db));
    let owner = DBOS::new(config("owner-instance-app", &db));

    let child = other
        .register_workflow("child", |()| async move { Ok::<u32, Error>(1) })
        .unwrap();
    let parent = {
        let child = child.clone();
        owner
            .register_workflow("parent", move |()| {
                let child = child.clone();
                async move { child.run(()).await }
            })
            .unwrap()
    };
    other.launch().await.expect("launch failed");
    owner.launch().await.expect("launch failed");

    let id = "borrows-another-instance";
    let error = parent
        .run_with(
            (),
            RunOptions {
                workflow_id: Some(id),
                ..RunOptions::default()
            },
        )
        .await
        .expect_err("the parent succeeded");
    assert!(
        matches!(&error, Error::WrongInstance { operation } if operation.contains("workflow")),
        "expected a wrong-instance refusal, got {error:?}"
    );

    // Refused before anything was written: no child, and no start on the parent.
    let reader = reader(&db).await;
    assert!(
        reader
            .get_workflow(&format!("{id}-0"))
            .await
            .expect("read failed")
            .is_none(),
        "no child row"
    );
    assert!(
        reader
            .list_workflow_steps(id, false, None, None, None)
            .await
            .expect("read failed")
            .is_empty(),
        "and no start recorded against the parent"
    );

    owner.shutdown().await;
    other.shutdown().await;
}

/// Awaiting a child from inside a step checkpoints nothing of its own, and is not an error.
///
/// The asymmetry with *starting* a child is deliberate, and both halves follow from a step being a
/// leaf. A start inside a step allocates ids that would shift every later step onto the wrong
/// replay slot, and there is no undurable version of it to fall back to, so it raises
/// [`Error::InsideStep`]. An await has such a version: the enclosing step's own checkpoint already
/// stands for whatever its body did, including the waiting, so the await simply runs plainly.
/// `DBOS::get_event` degrades in exactly the same way.
#[tokio::test]
async fn awaiting_a_child_inside_a_step_is_covered_by_that_step() {
    let db = test_database().await;
    let dbos = DBOS::new(config("await-in-step-app", &db));
    let child = dbos
        .register_workflow("child", |()| async move { Ok::<u32, Error>(41) })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", move |()| {
            let child = child.clone();
            async move {
                // Started at a step boundary, where a start has to happen...
                let handle = child.start(()).await?;
                // ...and awaited from inside a step, where there is no id to allocate. The handle
                // is not `Clone` and `result` consumes it, so it reaches the retryable closure
                // through a slot it takes from once.
                let slot = Arc::new(std::sync::Mutex::new(Some(handle)));
                let value = dbos::step("collect", move || {
                    let slot = Arc::clone(&slot);
                    async move {
                        let handle = slot.lock().unwrap().take().expect("the step runs once");
                        handle.result().await
                    }
                })
                .await?;
                Ok::<u32, Error>(value)
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let id = "collects-inside-a-step";
    assert_eq!(
        parent
            .run_with(
                (),
                RunOptions {
                    workflow_id: Some(id),
                    ..RunOptions::default()
                },
            )
            .await
            .expect("the parent failed"),
        41
    );

    let steps = reader(&db)
        .await
        .list_workflow_steps(id, true, None, None, None)
        .await
        .expect("read failed");
    assert_eq!(steps.len(), 2, "the start and the step: {steps:?}");
    assert_eq!(steps[0].step_name, "child", "the start");
    assert_eq!(
        steps[0].child_workflow_id.as_deref(),
        Some("collects-inside-a-step-0")
    );
    assert_eq!(
        steps[1].step_name, "collect",
        "the enclosing step, under its own name"
    );
    assert_eq!(
        steps[1].output.as_deref(),
        Some("41"),
        "which carries the child's value"
    );
    assert!(
        steps.iter().all(|step| step.step_name != "DBOS.getResult"),
        "the await added no checkpoint of its own: {steps:?}"
    );

    dbos.shutdown().await;
}

/// A recorded await that belongs to a *different* workflow is refused, not adopted.
///
/// The other half of the start position's own check, and the reason both exist: `check_step`
/// compares the step name, which leaves open whose outcome the row actually holds. For a child
/// the two cannot disagree — the handle's id came out of the start row moments earlier — so this
/// plants the disagreement directly, which is what a parent awaiting a handle it did not itself
/// start could otherwise reach by changing its code.
#[tokio::test]
async fn a_recorded_await_of_another_workflow_is_refused() {
    let db = test_database().await;
    let dbos = DBOS::new(config("stale-await-app", &db));
    let child = dbos
        .register_workflow("child", |()| async move { Ok::<u32, Error>(1) })
        .unwrap();

    // The start happens first, so its row is on disk before the gate; the await is what waits.
    let (tx, rx) = tokio::sync::oneshot::channel::<()>();
    let gate = Arc::new(std::sync::Mutex::new(Some(rx)));
    let parent = dbos
        .register_workflow("parent", move |()| {
            let child = child.clone();
            let gate = Arc::clone(&gate);
            async move {
                let handle = child.start(()).await?;
                let rx = gate.lock().unwrap().take().expect("the parent runs once");
                let _ = rx.await;
                handle.result().await
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let id = "awaits-the-wrong-one";
    let handle = parent
        .start_with(
            (),
            StartOptions {
                workflow_id: Some(id),
                ..StartOptions::default()
            },
        )
        .await
        .expect("start failed");

    let reader = reader(&db).await;
    let deadline = std::time::Instant::now() + Duration::from_secs(20);
    loop {
        let steps = reader
            .list_workflow_steps(id, false, None, None, None)
            .await
            .expect("read failed");
        if !steps.is_empty() {
            break;
        }
        assert!(
            std::time::Instant::now() < deadline,
            "the parent never recorded its start"
        );
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    // An await recorded at the position this parent is about to reach, naming a workflow that is
    // not the one it holds a handle to.
    reader
        .record_child_result(
            id,
            1,
            "somebody-elses-workflow",
            Outcome::Output(Some("7")),
            None,
            None,
            &owner_token(&reader, id).await,
        )
        .await
        .expect("planting the await failed");
    tx.send(()).expect("the parent is waiting on the gate");

    let error = handle.result().await.expect_err("the parent succeeded");
    match &error {
        Error::SystemDatabase(dbos::sysdb::Error::UnexpectedStep {
            step_id,
            expected,
            recorded,
            ..
        }) => {
            assert_eq!(*step_id, 1, "the await position");
            assert!(
                expected.contains("awaits-the-wrong-one-0"),
                "says which workflow it was awaiting: {expected}"
            );
            assert!(
                recorded.contains("somebody-elses-workflow"),
                "and whose outcome it found: {recorded}"
            );
        }
        other => panic!("expected an unexpected-step refusal, got {other:?}"),
    }

    dbos.shutdown().await;
}

/// **A parent may start a child that fails differently from it**, which is the case a start's own
/// error channel exists for.
///
/// What stands between the two channels is a *declaration* rather than a conversion: a start can
/// only fail in the engine's terms — nothing the child's application wrote has run yet — so
/// [`PendingStart::lift`] re-declares which channel this call reports in, before the await, and
/// `?` takes it from there. Without it the start would answer as the child fails, and a parent
/// that fails otherwise could not report a start failure at all: `?` has no conversion between two
/// application types, and `Error::lift` starts from the engine's channel.
///
/// The child's *own* failure is the other half, and it stays the handle's: that one really is an
/// application error, so the parent writes out what it makes of it.
///
/// A start that genuinely fails — attempted inside a step, where no durable start exists — is the
/// third: the refusal is the engine's, and it has to come back in the parent's channel rather than
/// the child's, which is the declaration being paid for at the poll rather than only made.
#[tokio::test]
async fn a_child_that_fails_differently_is_started_through_lift() {
    #[derive(Debug, thiserror::Error, serde::Serialize, serde::Deserialize)]
    #[error("the shipment was refused")]
    struct Refused;

    #[derive(Debug, thiserror::Error, serde::Serialize, serde::Deserialize)]
    #[error("billing gave up")]
    struct GaveUp;

    let db = test_database().await;
    let dbos = DBOS::new(config("mixed-channels-app", &db));
    let ship = dbos
        .register_workflow("ship", |()| async {
            Err::<u32, Error<Refused>>(Error::Application(Refused))
        })
        .unwrap();
    let bill = dbos
        .register_workflow("bill", {
            let ship = ship.clone();
            move |()| {
                let ship = ship.clone();
                async move {
                    // The whole of the test is that this line compiles: `lift` puts the start in
                    // the parent's channel, so `?` reaches for `Error<GaveUp>` and finds it.
                    let handle = ship.start(()).lift().await?;
                    // The child's error is the handle's and crosses no channel on its own — the
                    // parent decides what it means, which is what the type made it say.
                    let refused = matches!(handle.result().await, Err(Error::Application(Refused)));
                    if !refused {
                        Err(GaveUp)?;
                    }
                    // A start that actually fails, so the lift is paid rather than only declared:
                    // inside a step there is no durable start to make, and the engine's refusal
                    // still has to arrive in the parent's channel — `Error<GaveUp>` here — which
                    // is the conversion `PendingStart`'s poll does on the way out.
                    let refused_start = dbos::step::<(), GaveUp, _, _>("tries_to_start", || {
                        let ship = ship.clone();
                        async move { ship.start(()).lift::<GaveUp>().await.map(|_| ()) }
                    })
                    .await;
                    assert!(
                        matches!(refused_start, Err(Error::InsideStep { .. })),
                        "expected an inside-a-step refusal in the parent's channel, \
                         got {refused_start:?}"
                    );
                    Ok::<_, Error<GaveUp>>(refused)
                }
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    assert!(
        bill.run(()).await.expect("the parent failed"),
        "the parent started a child of another error type and read its refusal back"
    );

    dbos.shutdown().await;
}

/// **A race across a step and the await of a child**, which is the pairing `select_step!` exists
/// for: the child is started outside the race, and what races is the handle's `result` — a call
/// that *observes* rather than one that creates.
///
/// The await wins here, the losing step records nothing, and the race records the winner's
/// position. The ids show the whole shape: the start keeps its build-order id, the loser spends
/// one and writes nothing, and the select's own id follows its branches.
#[tokio::test]
async fn a_select_step_races_a_step_against_a_childs_result() {
    let db = test_database().await;
    let dbos = DBOS::new(config("race-child-app", &db));
    let child = dbos
        .register_workflow("child", |()| async move { Ok::<u32, Error>(7) })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", move |()| {
            let child = child.clone();
            async move {
                let started = child.start(()).await?;
                let outcome: u32 = dbos::select_step! {
                    // Never finishes, so the await wins however long the child's row takes to
                    // settle: a loser that merely slept would be racing a wall clock against the
                    // winner's database round trips.
                    slow = dbos::step("slow", || async {
                        std::future::pending::<()>().await;
                        Ok::<u32, Error>(0)
                    }) => slow?,
                    finished = started.result() => finished?
                }?;
                Ok::<u32, Error>(outcome)
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    let outcome = parent
        .run_with(
            (),
            RunOptions {
                workflow_id: Some("raced"),
                ..RunOptions::default()
            },
        )
        .await
        .expect("the parent failed");
    assert_eq!(
        outcome, 7,
        "the await won and its arm produced the child's answer"
    );

    let steps = reader(&db)
        .await
        .list_workflow_steps("raced", false, None, None, None)
        .await
        .expect("read failed");
    let positions: Vec<_> = steps
        .iter()
        .map(|step| {
            (
                step.step_id,
                step.step_name.as_str(),
                step.child_workflow_id.as_deref(),
            )
        })
        .collect();
    assert_eq!(
        positions,
        [
            (0, "child", Some("raced-0")),
            // Id 1 is the losing step, built and dropped without a row.
            (2, "DBOS.getResult", Some("raced-0")),
            (3, "DBOS.selectStep", None),
        ],
        "the await keeps its build-order id and the race records the winner behind it"
    );

    dbos.shutdown().await;
}

/// **A control signal winning a race is not the race's decision.** A step that ends in a
/// cancellation, an interruption or a database failure records nothing, so the workflow stays
/// pending and is recovered — and the race it won has to do the same. Recording the branch as the
/// winner would pin every recovery to a branch that never ran its body, and never race the other
/// again.
#[tokio::test]
async fn a_control_signal_winning_a_select_step_records_no_winner() {
    let db = test_database().await;
    let dbos = DBOS::new(config("control-race-app", &db));
    let raced = Arc::new(tokio::sync::Notify::new());
    let parent = {
        let raced = Arc::clone(&raced);
        dbos.register_workflow("parent", move |()| {
            let raced = Arc::clone(&raced);
            async move {
                // The arms hand the branch's own result out rather than `?`-ing it, so the
                // workflow reaches the notify whichever way the race went.
                let outcome: Result<Result<u32, Error>, Error> = dbos::select_step! {
                    // The shape a cancelled or interrupted body reports in: a control signal,
                    // which the step returns without checkpointing.
                    interrupted = dbos::step("interrupted", || async {
                        Err::<u32, Error>(Error::Interrupted {
                            workflow_id: "raced".to_owned(),
                        })
                    }) => interrupted,
                    // Never finishes, so the interrupted branch wins whatever its own round trip
                    // costs.
                    slow = dbos::step("slow", || async {
                        std::future::pending::<()>().await;
                        Ok::<u32, Error>(1)
                    }) => slow,
                };
                raced.notify_one();
                outcome.and_then(|won| won)
            }
        })
        .unwrap()
    };
    dbos.launch().await.expect("launch failed");

    let handle = parent
        .start_with(
            (),
            StartOptions {
                workflow_id: Some("raced"),
                ..StartOptions::default()
            },
        )
        .await
        .expect("the parent failed to start");
    tokio::time::timeout(Duration::from_secs(5), raced.notified())
        .await
        .expect("the race did not finish");

    let steps: Vec<_> = reader(&db)
        .await
        .list_workflow_steps("raced", false, None, None, None)
        .await
        .expect("read failed")
        .iter()
        .map(|step| (step.step_id, step.step_name.clone()))
        .collect();
    assert_eq!(
        steps,
        [],
        "neither the interrupted step nor the race it won left a row"
    );
    drop(handle);

    dbos.shutdown().await;
}

/// **A losing step's cancellation token fires**, as it does for a timeout, so work the runtime
/// cannot stop by dropping the future — a blocking thread, a task the body spawned — learns that
/// its step is over.
///
/// The winner waits for the loser to say it has registered a watcher rather than for a sleep to
/// expire, which is what makes the ordering a fact rather than a race against the database.
#[tokio::test]
async fn a_losing_step_has_its_cancellation_token_fired() {
    let db = test_database().await;
    let dbos = DBOS::new(config("loser-cancelled-app", &db));
    let released = Arc::new(tokio::sync::Notify::new());
    let workflow = {
        let released = Arc::clone(&released);
        dbos.register_workflow("workflow", move |()| {
            let released = Arc::clone(&released);
            async move {
                let (watching, watching_rx) = tokio::sync::oneshot::channel::<()>();
                let mut watching = Some(watching);
                let mut watching_rx_slot = Some(watching_rx);
                let outcome: u32 = dbos::select_step! {
                    fast = dbos::step("fast", move || {
                        // Taken by the first attempt; a step is built once and run once here.
                        let watching_rx = watching_rx_slot.take();
                        async move {
                            if let Some(watching_rx) = watching_rx {
                                let _ = watching_rx.await;
                            }
                            Ok::<u32, Error>(1)
                        }
                    }) => fast?,
                    slow = dbos::step("slow", move || {
                        let released = Arc::clone(&released);
                        let watching = watching.take();
                        async move {
                            let token = dbos::cancellation_token();
                            tokio::spawn(async move {
                                token.cancelled().await;
                                released.notify_one();
                            });
                            // Said only once the watcher is registered, and never withdrawn: from
                            // here the loser can be dropped and still be heard.
                            if let Some(watching) = watching {
                                let _ = watching.send(());
                            }
                            std::future::pending::<()>().await;
                            Ok::<u32, Error>(2)
                        }
                    }) => slow?,
                }?;
                Ok::<u32, Error>(outcome)
            }
        })
        .unwrap()
    };
    dbos.launch().await.expect("launch failed");

    let outcome = workflow.run(()).await.expect("the workflow failed");
    assert_eq!(outcome, 1, "the fast step won");
    tokio::time::timeout(Duration::from_secs(5), released.notified())
        .await
        .expect("the losing step's token never fired");

    dbos.shutdown().await;
}

/// **A workflow enqueued by name from inside a workflow is that workflow's child**: named for the
/// parent and its step, recorded at that step under the workflow's name, and found again by a
/// replay rather than enqueued a second time.
#[tokio::test]
async fn a_workflow_enqueued_by_name_is_a_child_and_a_replay_does_not_enqueue_it_again() {
    let db = test_database().await;
    let dbos = DBOS::new(config("by-name-app", &db));
    dbos.register_workflow("worker", |n: u32| async move { Ok::<u32, Error>(n * 2) })
        .unwrap();
    let parent = {
        let instance = dbos.clone();
        dbos.register_workflow("parent", move |n: u32| {
            let instance = instance.clone();
            async move {
                // By name: the parent holds no reference to the worker's registration.
                let handle: dbos::WorkflowHandle<u32> =
                    instance.enqueue("worker", "by-name-queue", n).await?;
                handle.result().await
            }
        })
        .unwrap()
    };
    dbos.launch().await.expect("launch failed");
    dbos.register_queue(
        "by-name-queue",
        QueueOptions::default(),
        QueueConflict::UpdateIfLatestVersion,
    )
    .await
    .expect("registration failed");

    let parent_id = "by-name-parent";
    let doubled = parent
        .run_with(
            21,
            RunOptions {
                workflow_id: Some(parent_id),
                ..RunOptions::default()
            },
        )
        .await
        .expect("the parent failed");
    assert_eq!(doubled, 42);

    let reader = reader(&db).await;
    let child = reader
        .get_workflow("by-name-parent-0")
        .await
        .unwrap()
        .expect("the child is named for its parent and step");
    assert_eq!(child.name.as_deref(), Some("worker"));
    assert_eq!(child.parent_workflow_id.as_deref(), Some(parent_id));
    assert_eq!(child.queue_name.as_deref(), Some("by-name-queue"));
    let steps = reader
        .list_workflow_steps(parent_id, true, None, None, None)
        .await
        .unwrap();
    assert_eq!(steps[0].step_name, "worker");
    assert_eq!(
        steps[0].child_workflow_id.as_deref(),
        Some("by-name-parent-0")
    );

    // Forked past the enqueue, so the replay meets the recorded start and joins its child.
    let replayed = dbos
        .fork::<u32, dbos::EngineOnly>(parent_id, 1)
        .await
        .expect("fork failed")
        .result()
        .await
        .expect("the replay failed");
    assert_eq!(replayed, 42);
    let workers = reader
        .list_workflows(
            &dbos::sysdb::types::WorkflowFilter {
                names: vec!["worker"],
                ..Default::default()
            },
            None,
        )
        .await
        .unwrap();
    assert_eq!(workers.len(), 1, "the replay enqueued again: {workers:?}");

    dbos.shutdown().await;
}

/// **Outside a workflow, an enqueue by name writes the row a client's enqueue writes**: this
/// application's unless another is named, and with no version unless one is named. From inside a
/// step it is refused, as a start is.
#[tokio::test]
async fn an_enqueue_by_name_writes_a_clients_row_and_is_refused_inside_a_step() {
    let db = test_database().await;
    let dbos = DBOS::new(config("by-name-row-app", &db));
    let in_step = {
        let instance = dbos.clone();
        dbos.register_workflow("in_step", move |()| {
            let instance = instance.clone();
            async move {
                dbos::step("enqueues", || {
                    let instance = instance.clone();
                    async move {
                        let _: dbos::WorkflowHandle<()> =
                            instance.enqueue("remote", "idle-queue", ()).await?;
                        Ok::<_, Error>(())
                    }
                })
                .await
            }
        })
        .unwrap()
    };
    dbos.launch().await.expect("launch failed");

    // A queue nothing polls, so the rows stay as they were written.
    let _: dbos::WorkflowHandle<u32> = dbos
        .enqueue_with(
            "remote",
            1u32,
            dbos::EnqueueOptions {
                workflow_id: Some("own-app"),
                ..dbos::EnqueueOptions::new("idle-queue")
            },
        )
        .await
        .expect("enqueue failed");
    let _: dbos::WorkflowHandle<u32> = dbos
        .enqueue_with(
            "remote",
            2u32,
            dbos::EnqueueOptions {
                workflow_id: Some("other-app"),
                app_name: Some("by-name-other-app"),
                app_version: Some("v9"),
                ..dbos::EnqueueOptions::new("idle-queue")
            },
        )
        .await
        .expect("enqueue failed");

    let reader = reader(&db).await;
    let own = reader.get_workflow("own-app").await.unwrap().unwrap();
    assert_eq!(own.status, WorkflowStatus::Enqueued);
    assert_eq!(own.name.as_deref(), Some("remote"));
    assert_eq!(own.app_name.as_deref(), Some("by-name-row-app"));
    assert_eq!(own.app_version, None, "no version unless one is named");
    assert_eq!(own.parent_workflow_id, None);
    let other = reader.get_workflow("other-app").await.unwrap().unwrap();
    assert_eq!(other.app_name.as_deref(), Some("by-name-other-app"));
    assert_eq!(other.app_version.as_deref(), Some("v9"));

    let refused = in_step.run(()).await.unwrap_err();
    assert!(matches!(refused, Error::InsideStep { .. }), "{refused:?}");
    let empty: dbos::Result<dbos::WorkflowHandle<()>> = dbos.enqueue("", "idle-queue", ()).await;
    assert!(
        matches!(empty, Err(Error::InvalidArgument { .. })),
        "{empty:?}"
    );

    dbos.shutdown().await;
}

/// **A deduplicated enqueue by name from a workflow is recorded either way it goes**: a join of the
/// holder is recorded at the parent's step and replays to the same holder, and a refusal is
/// recorded as one and replays as one, even once the holder is gone and the key is free.
#[tokio::test]
async fn a_deduplicated_enqueue_by_name_replays_the_way_it_went() {
    let db = test_database().await;
    let dbos = DBOS::new(config("by-name-dedup-app", &db));
    let parent = {
        let instance = dbos.clone();
        dbos.register_workflow("dedup_parent", move |policy: String| {
            let instance = instance.clone();
            async move {
                let duplication_policy = match policy.as_str() {
                    "join" => DuplicationPolicy::ReturnExisting,
                    _ => DuplicationPolicy::Reject,
                };
                let enqueued: dbos::Result<dbos::WorkflowHandle<()>> = instance
                    .enqueue_with(
                        "remote",
                        (),
                        dbos::EnqueueOptions {
                            deduplication_id: Some("held-key"),
                            duplication_policy,
                            ..dbos::EnqueueOptions::new("idle-dedup-queue")
                        },
                    )
                    .await;
                match enqueued {
                    Ok(handle) => Ok::<_, Error>(format!("joined {}", handle.workflow_id())),
                    Err(Error::SystemDatabase(dbos::sysdb::Error::QueueDeduplicated {
                        ..
                    })) => Ok("deduplicated".to_owned()),
                    Err(e) => Err(e),
                }
            }
        })
        .unwrap()
    };
    dbos.launch().await.expect("launch failed");

    // A queue nothing polls, so the holder keeps the key for as long as the test needs it.
    let _: dbos::WorkflowHandle<()> = dbos
        .enqueue_with(
            "remote",
            (),
            dbos::EnqueueOptions {
                workflow_id: Some("key-holder"),
                deduplication_id: Some("held-key"),
                ..dbos::EnqueueOptions::new("idle-dedup-queue")
            },
        )
        .await
        .expect("enqueue failed");

    let run = |id: &'static str, policy: &str| {
        parent.run_with(
            policy.to_owned(),
            RunOptions {
                workflow_id: Some(id),
                ..RunOptions::default()
            },
        )
    };
    assert_eq!(run("joins", "join").await.unwrap(), "joined key-holder");
    assert_eq!(run("refused", "reject").await.unwrap(), "deduplicated");
    let reader = reader(&db).await;
    let steps = reader
        .list_workflow_steps("joins", true, None, None, None)
        .await
        .unwrap();
    assert_eq!(steps[0].child_workflow_id.as_deref(), Some("key-holder"));

    let replay = |id: &'static str| {
        let dbos = dbos.clone();
        async move {
            dbos.fork::<String, Error>(id, 1)
                .await
                .expect("fork failed")
                .result()
                .await
                .expect("the replay failed")
        }
    };
    assert_eq!(replay("joins").await, "joined key-holder");
    // The key is free now, so only the record can say this enqueue was refused.
    dbos.delete("key-holder").await.expect("delete failed");
    assert_eq!(replay("refused").await, "deduplicated");
    let remote = reader
        .list_workflows(
            &dbos::sysdb::types::WorkflowFilter {
                names: vec!["remote"],
                ..Default::default()
            },
            None,
        )
        .await
        .unwrap();
    assert!(remote.is_empty(), "a replay enqueued again: {remote:?}");

    dbos.shutdown().await;
}

/// **An enqueue by name from a workflow with a deadline carries that deadline**, as a child start
/// does, unless it is given a timeout of its own or told to run unbounded.
#[tokio::test]
async fn an_enqueue_by_name_inherits_its_parents_deadline() {
    let db = test_database().await;
    let dbos = DBOS::new(config("by-name-deadline-app", &db));
    let parent = {
        let instance = dbos.clone();
        dbos.register_workflow("deadline_parent", move |()| {
            let instance = instance.clone();
            async move {
                let _: dbos::WorkflowHandle<()> = instance
                    .enqueue("remote", "idle-deadline-queue", ())
                    .await?;
                let _: dbos::WorkflowHandle<()> = instance
                    .enqueue_with(
                        "remote",
                        (),
                        dbos::EnqueueOptions {
                            timeout: Timeout::Explicit(Duration::from_secs(3600)),
                            ..dbos::EnqueueOptions::new("idle-deadline-queue")
                        },
                    )
                    .await?;
                let _: dbos::WorkflowHandle<()> = instance
                    .enqueue_with(
                        "remote",
                        (),
                        dbos::EnqueueOptions {
                            timeout: Timeout::None,
                            ..dbos::EnqueueOptions::new("idle-deadline-queue")
                        },
                    )
                    .await?;
                Ok::<_, Error>(())
            }
        })
        .unwrap()
    };
    dbos.launch().await.expect("launch failed");

    parent
        .start_with(
            (),
            StartOptions {
                workflow_id: Some("deadline-parent"),
                timeout: Timeout::Explicit(Duration::from_secs(600)),
                ..StartOptions::default()
            },
        )
        .await
        .expect("start failed")
        .result()
        .await
        .expect("the parent failed");

    let reader = reader(&db).await;
    let parent_row = reader
        .get_workflow("deadline-parent")
        .await
        .unwrap()
        .unwrap();
    let inherited = reader
        .get_workflow("deadline-parent-0")
        .await
        .unwrap()
        .unwrap();
    let own = reader
        .get_workflow("deadline-parent-1")
        .await
        .unwrap()
        .unwrap();
    assert!(parent_row.deadline.is_some());
    assert_eq!(inherited.deadline, parent_row.deadline);
    assert_eq!(own.timeout, Some(Duration::from_secs(3600)));
    assert_ne!(own.deadline, parent_row.deadline);
    let unbounded = reader
        .get_workflow("deadline-parent-2")
        .await
        .unwrap()
        .unwrap();
    assert_eq!((unbounded.timeout, unbounded.deadline), (None, None));

    dbos.shutdown().await;
}

/// **An enqueue by name refuses an app name or version no executor could have, and an empty id or
/// key, before it takes a step id**, so the refusal costs the parent no step and leaves no record.
#[tokio::test]
async fn bad_enqueue_options_are_refused_before_the_step_id_is_taken() {
    let db = test_database().await;
    let dbos = DBOS::new(config("by-name-refused-app", &db));
    let parent = {
        let instance = dbos.clone();
        dbos.register_workflow("refusing_parent", move |()| {
            let instance = instance.clone();
            async move {
                let mut refusals = Vec::new();
                for options in [
                    dbos::EnqueueOptions {
                        app_name: Some("Not-An-App"),
                        ..dbos::EnqueueOptions::new("idle-queue")
                    },
                    dbos::EnqueueOptions {
                        app_version: Some(""),
                        ..dbos::EnqueueOptions::new("idle-queue")
                    },
                    dbos::EnqueueOptions {
                        workflow_id: Some(""),
                        ..dbos::EnqueueOptions::new("idle-queue")
                    },
                    dbos::EnqueueOptions {
                        deduplication_id: Some(""),
                        ..dbos::EnqueueOptions::new("idle-queue")
                    },
                    dbos::EnqueueOptions {
                        partition_key: Some(""),
                        ..dbos::EnqueueOptions::new("idle-queue")
                    },
                ] {
                    let refused: dbos::Result<dbos::WorkflowHandle<()>> =
                        instance.enqueue_with("remote", (), options).await;
                    match refused {
                        Err(Error::InvalidArgument { detail, .. }) => refusals.push(detail),
                        other => panic!("not refused: {:?}", other.map(|_| ())),
                    }
                }
                dbos::step("after", || async { Ok::<_, Error>(()) }).await?;
                Ok::<_, Error>(refusals)
            }
        })
        .unwrap()
    };
    dbos.launch().await.expect("launch failed");

    let refusals = parent
        .run_with(
            (),
            RunOptions {
                workflow_id: Some("refusing-parent"),
                ..RunOptions::default()
            },
        )
        .await
        .expect("the parent failed");
    assert!(refusals[0].contains("`app_name`"), "{refusals:?}");
    assert!(refusals[1].contains("`app_version`"), "{refusals:?}");
    assert!(refusals[2].contains("`workflow_id`"), "{refusals:?}");
    assert!(refusals[3].contains("`deduplication_id`"), "{refusals:?}");
    assert!(refusals[4].contains("`partition_key`"), "{refusals:?}");
    let steps = reader(&db)
        .await
        .list_workflow_steps("refusing-parent", true, None, None, None)
        .await
        .unwrap();
    assert_eq!(steps.len(), 1, "{steps:?}");
    assert_eq!(
        (steps[0].step_id, steps[0].step_name.as_str()),
        (0, "after")
    );

    dbos.shutdown().await;
}

/// A registered workflow that captures a `WorkflowRef` to start children does not keep the
/// instance alive.
///
/// The registry that owns the closure lives on the instance, so a reference that held the instance
/// would be a cycle and the instance, its executor and its connection pool would never be freed.
/// The probe is captured beside the reference: the closure is dropped only with the registry, and
/// the registry only with the instance, so the probe's going is the instance's. Every form a
/// workflow body starts a child through is exercised — start, run and debounce — since each one
/// reaches the executor through the reference.
#[tokio::test]
async fn a_captured_workflow_ref_does_not_keep_the_instance_alive() {
    let db = test_database().await;
    let dbos = DBOS::new(config("captured-ref-app", &db));
    let probe = Arc::new(());
    let freed = Arc::downgrade(&probe);

    let child = dbos
        .register_workflow("child", |n: u32| async move { Ok::<u32, Error>(n * 2) })
        .unwrap();
    let parent = dbos
        .register_workflow("parent", move |n: u32| {
            let child = child.clone();
            let _probe = Arc::clone(&probe);
            async move {
                let started = child.start(n).await?.result().await?;
                let ran = child.run(n + 1).await?;
                let debounced = Debouncer::new(&child)
                    .debounce("captured", Duration::from_millis(10), n + 2)
                    .await?
                    .result()
                    .await?;
                Ok::<u32, Error>(started + ran + debounced)
            }
        })
        .unwrap();
    dbos.launch().await.expect("launch failed");

    assert_eq!(parent.run(1).await.expect("the parent failed"), 2 + 4 + 6);

    dbos.shutdown().await;
    drop(dbos);
    assert!(
        freed.upgrade().is_none(),
        "the instance is still alive: something holds it after its last handle was dropped"
    );

    // A reference that outlived its instance has nothing to start through: the slot it holds was
    // emptied by shutdown.
    let error = parent
        .start(1)
        .await
        .expect_err("started on a dropped instance");
    assert!(
        matches!(&error, Error::NotLaunched { .. }),
        "expected not-launched, got {error:?}"
    );
}