dbos 0.6.0

//! The instance and the executor: the two halves of the public lifecycle.

use std::sync::{Arc, RwLock};
use std::time::{Duration, Instant};

use tokio_util::sync::CancellationToken;

use crate::config::Config;
use crate::connection::Connection;
use crate::identity::{Environment, Identity};
use crate::registry::{Registry, Snapshot};
use crate::serialization::Serializer;
use crate::sysdb::SystemDatabase;
use crate::workflow::{Role, Tasks};
use crate::{Error, Result};

/// The running half of DBOS, created by [`DBOS::launch`] and dropped by [`DBOS::shutdown`].
///
/// Everything that needs a database or a runtime lives here rather than on the instance, so that
/// "not launched" is one absent value rather than a scatter of `Option`s.
///
/// **The connection is held rather than embedded**, which is the split a [`Client`](crate::Client)
/// rests on. Every other field here exists because this process *runs* workflows: the id, version
/// and application name a claim stamps, the runtime they are spawned onto, the registry snapshot
/// they are resolved through, the task sets shutdown reaches. A client runs none, so it holds the
/// connection alone, and everything both surfaces do is a method on that rather than on this
/// type.
pub struct Executor {
    conn: Arc<Connection>,
    executor_id: String,
    app_version: String,
    app_name: String,
    app_id: String,
    runtime: crate::rt::Spawner,
    workflows: Snapshot,
    listen_queues: Option<Vec<String>>,
    /// One set per [`Role`], because shutdown stops them in turn: see [`stop`](Self::stop).
    producer_tasks: Tasks,
    workflow_tasks: Tasks,
    monitor_tasks: Tasks,
    /// Cancelled when shutdown begins, which is what the producers' loops watch to stop.
    stopping: CancellationToken,
}

impl Executor {
    /// Resolves this executor's identity, connects, and registers the running version.
    ///
    /// The mirror of [`shutdown`](Self::shutdown), and here for the same reason: these are the
    /// executor's own fields, so the executor is what sets them up. [`DBOS::launch`] is left with
    /// what is actually its business — deciding whether to build one at all, and installing it.
    ///
    /// Also **recovers**, returning the ids it re-enqueued. Done *here*, before launch returns,
    /// because what counts as abandoned must be decided before the application can start anything:
    /// a workflow started the instant launch returns is `PENDING` under this executor's id too,
    /// and a sweep running any later would tear it off its runner and offer it to the fleet. It is
    /// one write rather than a list of workflows to run, so nothing about it is backgrounded.
    async fn start(config: &Config, workflows: Snapshot) -> Result<(Self, Vec<String>)> {
        config.validate()?;

        if workflows.is_empty() {
            tracing::warn!(
                "no workflows are registered: this executor will recover nothing and dequeue \
                 nothing. Register before calling `launch`."
            );
        }

        // The name, the version and the executor id are the deployment's business as much as the
        // configuration's, and one place reconciles them.
        let Identity {
            app_name,
            app_version,
            executor_id,
            app_id,
        } = crate::identity::resolve(config, &Environment::read())?;

        let conn = Connection::for_application(config, &executor_id, &app_name).await?;

        // Everything that can fail after the connect goes through one call, so there is one error
        // path and it closes the handle. `connect` has already spawned the listener and the
        // notifier, and the listener's only way out of its loop is the pool closing — so a handle
        // dropped without `close` leaves both tasks and every connection alive for the life of the
        // process, with each retried `launch` adding another set.
        let recovered = match prepare(conn.sysdb(), &executor_id, &app_version).await {
            Ok(recovered) => recovered,
            Err(error) => {
                conn.close().await;
                return Err(error);
            }
        };

        tracing::info!(app_name, executor_id, app_version, "DBOS launched");

        let executor = Self {
            conn: Arc::new(conn),
            executor_id,
            app_version,
            app_id,
            // The runtime is the executor's, taken here. `start` is `async`, so some runtime is
            // running it; this panics only if it is one this build was not given a feature for.
            runtime: crate::rt::Spawner::current(),
            workflows,
            app_name,
            listen_queues: config.listen_queues.clone(),
            producer_tasks: Tasks::default(),
            workflow_tasks: Tasks::default(),
            monitor_tasks: Tasks::default(),
            stopping: CancellationToken::new(),
        };
        Ok((executor, recovered))
    }

    /// The connection this executor talks through, which is all a caller needs to read or write a
    /// row.
    ///
    /// Handed out rather than proxied so that everything both surfaces do can be a method on the
    /// connection and be called by a [`Client`](crate::Client) too — see that module.
    pub(crate) fn connection(&self) -> &Arc<Connection> {
        &self.conn
    }

    /// The system database this executor is running against.
    pub(crate) fn sysdb(&self) -> &dyn SystemDatabase {
        self.conn.sysdb()
    }

    /// Identifies this process among the executors sharing the database.
    pub fn executor_id(&self) -> &str {
        &self.executor_id
    }

    /// Names the application whose rows this executor owns.
    ///
    /// **Its own, and not an [`Option`].** The connection reports a name that may be absent,
    /// because the one a [`Client`](crate::Client) opens may speak for every application at once.
    /// An executor may not: the identity is resolved and validated at launch, before anything
    /// connects, and the name that wins — on DBOS Cloud the deployment's `DBOS_APP_NAME` rather
    /// than [`Config::app_name`](crate::Config::app_name) — is the ownership key on every row this
    /// process writes. Holding it here rather than unwrapping the database's is what keeps that a
    /// fact of the type instead of an invariant each caller has to remember.
    pub fn app_name(&self) -> &str {
        &self.app_name
    }

    /// The tasks of one role this executor started and has not seen finish.
    pub(crate) fn tasks(&self, role: Role) -> &Tasks {
        match role {
            Role::Producer => &self.producer_tasks,
            Role::Workflow => &self.workflow_tasks,
            Role::Monitor => &self.monitor_tasks,
        }
    }

    /// Sleeps for `duration` between a producer loop's passes, unless shutdown begins first.
    ///
    /// `true` when it slept the whole duration and the loop should make another pass; `false` once
    /// shutdown has begun, and at once rather than at the end of the sleep. The
    /// producers stop here, between passes, rather than wherever an abort would land: a dequeue
    /// whose claim has committed goes on to start what it claimed, so a graceful shutdown drains
    /// those workflows instead of leaving them `PENDING` and unstarted.
    pub(crate) async fn sleep_unless_stopping(&self, duration: Duration) -> bool {
        // Biased, with the token first: a wait that has already elapsed — a schedule catching up
        // waits for nothing — would otherwise win half the time and run one more pass.
        tokio::select! {
            biased;
            () = self.stopping.cancelled() => false,
            () = crate::rt::sleep(duration) => true,
        }
    }

    /// Whether shutdown has begun, for a producer about to start a pass.
    pub(crate) fn is_stopping(&self) -> bool {
        self.stopping.is_cancelled()
    }

    /// Which queues this executor dequeues from, or `None` for all of them.
    ///
    /// Captured at launch: it is configuration, not a knob. The *set* it narrows is still rebuilt
    /// from the table on every sweep, which is where the dynamic behaviour lives.
    pub(crate) fn listen_queues(&self) -> Option<&[String]> {
        self.listen_queues.as_deref()
    }

    /// How often an adopting caller asks whether the run that won has finished.
    pub(crate) fn outcome_poll_interval(&self) -> std::time::Duration {
        self.conn.outcome_poll_interval()
    }

    /// The version of the application's code, as workflow rows record it.
    ///
    /// **`app_` here, `application_` in the rows `sysdb` writes**, and the two meet at exactly
    /// this call: the API names the pair [`app_name`](Self::app_name) and `app_version`, matching
    /// [`APP_VERSION_ENV`](crate::APP_VERSION_ENV) and the `DBOS__APPVERSION` every DBOS SDK
    /// reads — and so do the `sysdb` types the API hands back, such as
    /// [`WorkflowRecord`](crate::sysdb::types::WorkflowRecord) — while the `sysdb` types only the
    /// engine writes spell them `application_name` and `application_version` because those are
    /// the columns' own names. The conversion happens where a value crosses into such a type —
    /// `NewWorkflow { application_version: Some(executor.app_version()), .. }` — which is the
    /// same boundary every other schema word stops at.
    pub fn app_version(&self) -> &str {
        &self.app_version
    }

    /// This application's DBOS Cloud id, empty when the process is not running on DBOS Cloud.
    ///
    /// Read from [`APP_ID_ENV`](crate::APP_ID_ENV) and from nowhere else: an id is issued to a
    /// deployment rather than chosen by an application, so there is no configuration field for it.
    pub fn app_id(&self) -> &str {
        &self.app_id
    }

    /// How payloads are encoded.
    pub(crate) fn serializer(&self) -> &Serializer {
        self.conn.serializer()
    }

    /// The workflows this executor was launched with.
    ///
    /// A snapshot, frozen at launch. Recovery reads it while the application may still be calling
    /// into the instance, and a map that could grow underneath a recovery pass is a recovery pass
    /// that may or may not find a workflow depending on timing.
    pub(crate) fn workflows(&self) -> &Snapshot {
        &self.workflows
    }

    /// The runtime workflows are spawned onto.
    ///
    /// Captured at launch rather than taken from whoever is calling, for two reasons that both
    /// come down to ownership. Recovery spawns from a background task rather than from user code,
    /// so "the caller's runtime" is not one answer. And shutdown has to abort exactly the tasks
    /// this executor started, which needs a handle it chose rather than one it inherited.
    pub(crate) fn runtime(&self) -> &crate::rt::Spawner {
        &self.runtime
    }

    /// Stops this executor starting workflows, waits up to `workflow_completion_timeout` for those
    /// running to finish, and cancels whatever is still running at the deadline.
    ///
    /// The first half of shutdown; [`close`](Self::close) is the second, and the instance stops
    /// holding the executor between the two. Here rather than in [`DBOS::shutdown_with`] because
    /// this is where the things being stopped live, and teardown that grows a step at a time
    /// wants one place to grow.
    ///
    /// **The producers stop first**, so the set being drained cannot grow underneath the drain: a
    /// dequeue that had already claimed work when shutdown began starts it, and that workflow is
    /// drained with the rest. A workflow started some other way — by the application, or as a
    /// running workflow's child — is waited for if it starts while the drain is still waiting,
    /// which is what lets a parent finish. One that starts after the drain has ended is cancelled
    /// as it arrives and stays `PENDING`. A child a workflow *enqueues* is not dequeued here, so a
    /// parent waiting on one waits out the timeout.
    ///
    /// **A zero timeout waits for nothing**, and is the shutdown this crate always had: the
    /// producers are stopped, every workflow is cancelled, and the rows stay `PENDING` for
    /// recovery. A non-zero one cancels the same way at its deadline, so a workflow that overruns
    /// it is recovered from its last checkpoint rather than marked. What keeps an interrupted
    /// workflow from being *recorded* as failed is not this ordering but `Error::control`, which
    /// treats a database failure as a signal rather than an outcome.
    ///
    /// Takes `&self`, not `self`: a workflow still running holds another handle, and shutting down
    /// is precisely the moment when that is true.
    pub(crate) async fn stop(&self, workflow_completion_timeout: Duration) {
        self.stopping.cancel();
        let draining = !workflow_completion_timeout.is_zero();
        if draining {
            let deadline = crate::rt::now() + workflow_completion_timeout;
            // Waited on before the drain, not alongside it: a dequeue mid-dispatch is about to add
            // to the set the drain is watching, and a drain that saw it empty first would finish
            // before that workflow started.
            if self.producer_tasks.wait_all_until(deadline).await {
                self.drain(deadline).await;
            }
        }

        // Cancel, wait, then report — and the wait is what the ordering rests on. `abort` only
        // schedules cancellation at a task's next yield point, so without waiting, the system
        // database could close under bodies that were still running, and returning would mean
        // "told to stop" rather than "stopped". Waiting is also what lets a relaunch in this
        // process start against a genuinely quiet executor.
        let producers = self.producer_tasks.abort_all().await;
        let abandoned = self.workflow_tasks.abort_all().await;
        if draining {
            if !producers.is_empty() {
                tracing::warn!(
                    tasks = ?producers,
                    "the dequeue loop or the scheduler was still mid-pass at the shutdown \
                     deadline and was cancelled"
                );
            }
            if !abandoned.is_empty() {
                tracing::warn!(
                    workflows = abandoned.len(),
                    workflow_ids = ?abandoned,
                    "workflows still running at the shutdown deadline were cancelled; they stay \
                     PENDING and are recovered from their last checkpoint"
                );
            }
        } else if !abandoned.is_empty() {
            tracing::info!(
                workflows = abandoned.len(),
                workflow_ids = ?abandoned,
                "cancelled workflows still running; they stay PENDING"
            );
        }
    }

    /// Waits for the running workflows to finish on their own, up to `deadline`, saying once a
    /// second which ones it is still waiting for.
    ///
    /// Waits only. Nothing is interrupted here, at the deadline or before it: what is left is
    /// cancelled by [`stop`](Self::stop), after this returns.
    async fn drain(&self, deadline: Instant) {
        loop {
            let report_at = (crate::rt::now() + DRAIN_REPORT_INTERVAL).min(deadline);
            if self.workflow_tasks.wait_all_until(report_at).await || report_at >= deadline {
                return;
            }
            let running = self.workflow_tasks.running();
            tracing::info!(
                workflows = running.len(),
                workflow_ids = ?running,
                "waiting for workflows to finish before shutting down"
            );
        }
    }

    /// Stops the background monitors and closes the system database. Called after
    /// [`stop`](Self::stop), once no workflow is running.
    ///
    /// **The monitors outlive the drain** because a workflow still running may be waiting on the
    /// timeout sweep to record that its deadline has passed. **The system database closes last**,
    /// so everything before it could still checkpoint.
    ///
    /// A Conductor connection, once there is one, is closed between the two: after the drain, so
    /// the executor stays visibly alive to Conductor for as long as it is running workflows, and
    /// before the pool, which it reads through.
    pub(crate) async fn close(&self) {
        self.monitor_tasks.abort_all().await;
        self.conn.close().await;
    }
}

/// How often a draining shutdown says which workflows it is still waiting for.
const DRAIN_REPORT_INTERVAL: Duration = Duration::from_secs(1);

impl std::fmt::Debug for Executor {
    /// Hand-written because a `dyn SystemDatabase` is not `Debug` and should not become so: the
    /// trait exists to keep a driver type out of the engine, and a `Debug` bound would put one
    /// back in through the derive.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Executor")
            .field("executor_id", &self.executor_id)
            .field("app_version", &self.app_version)
            .finish_non_exhaustive()
    }
}

/// What [`DBOS::shutdown_with`] does with the workflows still running.
///
/// A struct rather than a bare argument, matching the other `_with` calls of this crate.
/// Deliberately **not** `#[non_exhaustive]`, which would forbid the `..Default::default()` form.
#[derive(Debug, Clone, Default)]
pub struct ShutdownOptions {
    /// How long to wait for the workflows running in this process to finish before cancelling
    /// them.
    ///
    /// Zero, the default, waits for nothing: every running workflow is cancelled at once. A
    /// cancelled workflow is not marked; its row stays `PENDING` and it is recovered from its last
    /// checkpoint, so the timeout trades shutdown time for work redone, never for work lost.
    pub workflow_completion_timeout: Duration,
}

/// A DBOS instance: the whole public surface of the library.
///
/// Cheap to clone — it is an `Arc` internally, so a clone is another handle to the same instance
/// rather than another instance. Clone it into whatever holds application state.
///
/// The instance owns the configuration and the workflow registry, holding them for the life of the
/// application. [`launch`](Self::launch) constructs an [`Executor`] and
/// [`shutdown`](Self::shutdown) drops it, so relaunching in one process is supported — which is
/// why this is an instance whose state changes rather than a builder that becomes a handle. A
/// type-state would make "not launched" a compile error and would also make relaunching
/// inexpressible.
///
/// ```no_run
/// # async fn f() -> dbos::Result<()> {
/// use dbos::{Config, DBOS};
///
/// async fn greet(name: String) -> dbos::Result<String> {
///     Ok(format!("Hello, {name}"))
/// }
///
/// let mut config = Config::from_env("my-app");
/// config.app_version = Some("0.1.0".to_owned());
/// let dbos = DBOS::new(config);
/// let greet = dbos.register_workflow("greet", greet)?;
/// dbos.launch().await?;
///
/// let greeting = greet.start("world".to_owned()).await?.result().await?;
///
/// dbos.shutdown().await;
/// # let _ = greeting;
/// # Ok(())
/// # }
/// ```
#[derive(Clone)]
pub struct DBOS(Arc<Inner>);

struct Inner {
    config: Config,
    /// Every workflow registered so far. Lives on the instance rather than the executor, because
    /// registration happens before there is an executor.
    registry: Registry,
    /// `None` until launched. Shared with every [`WorkflowRef`](crate::WorkflowRef) — see
    /// [`ExecutorSlot`].
    executor: Arc<ExecutorSlot>,
    /// Serializes launch against shutdown, and either against itself.
    ///
    /// **This guard is deliberately held across await points** — `Executor::start` connects and
    /// migrates — which is exactly why it is a `tokio` mutex and the slot above is a `std` one.
    /// The two are not interchangeable. A `std` guard is `!Send` and blocks the *thread*, so
    /// holding one across an await stops the future being `Send` and risks a worker thread
    /// blocking on a lock whose holder needs that same thread to make progress. A `tokio` guard
    /// parks the *task*, and is meant to be held.
    ///
    /// The rule this leaves is narrow and worth stating in those terms: no guard from the `std`
    /// `RwLock` above ever crosses an await, which `the_lifecycle_futures_are_send` checks by
    /// construction rather than by inspection.
    lifecycle: tokio::sync::Mutex<()>,
}

impl Drop for Inner {
    fn drop(&mut self) {
        // `shutdown` is async and `Drop` is not, so this can only report. Closing here would mean
        // blocking on a runtime from inside a drop, which deadlocks when the drop is itself on
        // that runtime.
        if self.executor.read().is_some() {
            tracing::warn!(
                app_name = self.config.app_name,
                "DBOS dropped while still launched: call `shutdown().await` to close the system \
                 database cleanly. Its connections are closed by the pool's own drop, but any \
                 workflow still running is left where it stood."
            );
        }
    }
}

impl DBOS {
    /// A new instance. Nothing is connected until [`launch`](Self::launch).
    pub fn new(config: Config) -> Self {
        Self(Arc::new(Inner {
            config,
            registry: Registry::default(),
            executor: Arc::default(),
            lifecycle: tokio::sync::Mutex::new(()),
        }))
    }

    /// The configuration this instance was built with.
    pub fn config(&self) -> &Config {
        &self.0.config
    }

    /// Everything registered so far.
    pub(crate) fn registry(&self) -> &Registry {
        &self.0.registry
    }

    /// Whether an executor is running.
    pub fn is_launched(&self) -> bool {
        self.read_executor().is_some()
    }

    /// Connects to the system database, migrates it, registers this application version, and starts
    /// the executor.
    ///
    /// **Idempotent**: launching an already-launched instance warns and returns. It is not an error
    /// because the common cause is two entry points both being defensive, and failing there would
    /// punish the caller who did nothing wrong.
    pub async fn launch(&self) -> Result<()> {
        let _lifecycle = self.0.lifecycle.lock().await;
        if self.read_executor().is_some() {
            tracing::warn!("DBOS was already launched");
            return Ok(());
        }

        // Freezing the registry and reading it are one act, so a registration cannot land between
        // the two and be handed back a `WorkflowRef` for a workflow this executor will not have.
        let workflows = self.0.registry.snapshot();
        let (executor, recovered) = match Executor::start(&self.0.config, workflows).await {
            Ok(started) => started,
            Err(error) => {
                // Nothing was installed, so nothing holds the snapshot and registration is open
                // again. Leaving it frozen would make one failed launch permanent.
                self.0.registry.thaw();
                return Err(error);
            }
        };
        let executor = Arc::new(executor);
        *self.write_executor() = Some(Arc::clone(&executor));
        // Recovery already happened, inside `Executor::start`: it is one write, and doing it before
        // launch returns is what keeps it clear of workflows the application starts next. What is
        // left is to start polling, which is also what will run the re-enqueued work.
        if !recovered.is_empty() {
            tracing::debug!(
                workflows = recovered.len(),
                "recovered workflows are on their queues"
            );
        }
        // Every executor fires every schedule of its application; the deterministic id of each run
        // is what keeps the fleet from enqueueing a tick more than once.
        crate::scheduler::spawn(
            Arc::clone(&executor),
            self.0.config.scheduler_polling_interval(),
        );
        // The dequeue loop starts unconditionally, because a queue this process never registered
        // is still one it should dequeue from: the worker set is rebuilt from the `queues` table
        // on every supervisor sweep, not from this instance's `register_queue` calls. A queue
        // another *process of this application* registered is therefore picked up, as is one
        // registered against this instance after launch, without a restart.
        //
        // Another *application's* queue is not, and that is the boundary: the search scopes to
        // this application's rows plus the unclaimed ones, so a peer application's backlog is
        // never taken. Sharing a backlog is what a fleet of one application does; reaching into
        // another's would be the same redirection that makes registering over its queue name an
        // error.
        // Every executor sweeps every deadline of its application, so a workflow nobody is running
        // still times out.
        crate::timeout_sweep::spawn(Arc::clone(&executor));
        crate::dequeue::spawn(executor);
        Ok(())
    }

    /// Stops the executor and closes the system database, cancelling any workflow still running.
    ///
    /// A workflow started while this runs is cancelled as it arrives, and its caller gets
    /// [`Error::Interrupted`].
    ///
    /// [`shutdown_with`](Self::shutdown_with) with the default options, which wait for nothing: a
    /// workflow running in this process is cancelled where it stands and stays `PENDING`, for a
    /// later executor to recover from its last checkpoint. To let running workflows finish first,
    /// give [`ShutdownOptions::workflow_completion_timeout`].
    ///
    /// **Idempotent**: shutting down an instance that is not launched does nothing.
    /// Relaunching afterwards is supported and is what test suites do.
    pub async fn shutdown(&self) {
        self.shutdown_with(ShutdownOptions::default()).await;
    }

    /// Stops the executor and closes the system database, first waiting up to
    /// [`workflow_completion_timeout`](ShutdownOptions::workflow_completion_timeout) for the
    /// workflows running in this process to finish.
    ///
    /// In order:
    ///
    /// 1. **The dequeue loop and the scheduler stop**, so no queued workflow starts here from now
    ///    on. A workflow started any other way — by the application, or as a running workflow's
    ///    child — still starts while step 2 is waiting, and is waited for with the rest. One
    ///    started after that is cancelled as it arrives, and its caller gets
    ///    [`Error::Interrupted`].
    /// 2. **Running workflows are given until the timeout to finish**, and are not interrupted
    ///    meanwhile. Which ones are still running is logged once a second.
    /// 3. **Whatever is still running at the deadline is cancelled**, and logged by id. Its row is
    ///    left `PENDING`, so it is recovered from its last checkpoint rather than marked failed.
    /// 4. The workflow-timeout sweep stops, after the wait, since a workflow still running may be
    ///    waiting on it.
    /// 5. The system database closes last, so a workflow finishing during the wait can still
    ///    checkpoint.
    ///
    /// The instance stays launched until step 3 is over, so a workflow finishing during the wait
    /// can still call into it.
    ///
    /// A workflow that waits on a child it *enqueued* waits out the timeout: the child is not
    /// dequeued here once shutdown has begun. So does a workflow that calls this itself, since it
    /// is one of the workflows being waited for; it is then cancelled and stays `PENDING`.
    ///
    /// **Idempotent**: shutting down an instance that is not launched does nothing, and a second
    /// call made while the first is waiting returns once the first has finished.
    ///
    /// **Not cancelled by dropping the future.** The teardown runs as a task of its own, and a
    /// caller that stops waiting — a `tokio::time::timeout` around this call, or a `select!` —
    /// leaves it to finish. Cut short, it would leave an instance that reports itself launched
    /// with its producers stopped and every new workflow cancelled as it arrives.
    ///
    /// ```no_run
    /// # async fn f(dbos: &dbos::DBOS) {
    /// use std::time::Duration;
    ///
    /// dbos.shutdown_with(dbos::ShutdownOptions {
    ///     workflow_completion_timeout: Duration::from_secs(30),
    /// })
    /// .await;
    /// # }
    /// ```
    pub async fn shutdown_with(&self, options: ShutdownOptions) {
        let dbos = self.clone();
        match crate::rt::spawn(async move { dbos.tear_down(options).await }).await {
            Ok(()) => {}
            Err(join) if join.is_panic() => std::panic::resume_unwind(join.into_panic()),
            // The runtime itself is shutting down, and the task with it.
            Err(_) => {}
        }
    }

    /// The whole of [`shutdown_with`](Self::shutdown_with), run as its own task.
    async fn tear_down(&self, options: ShutdownOptions) {
        let _lifecycle = self.0.lifecycle.lock().await;
        let Some(executor) = self.read_executor() else {
            return;
        };
        // Stopped while still installed: a workflow given time to finish may call back into this
        // instance — to send, to start a child — and an instance that already said it was not
        // launched would fail it with an error that its body could record as its outcome. Taken
        // once nothing is running that could make such a call.
        executor.stop(options.workflow_completion_timeout).await;
        self.write_executor().take();
        executor.close().await;
        // The instance outlives the executor, so registration opens again: relaunching takes a
        // fresh snapshot, and what that executor holds is its own.
        self.0.registry.thaw();
        tracing::info!(app_name = self.0.config.app_name, "DBOS shut down");
    }

    /// Identifies this process among the executors sharing the database.
    pub fn executor_id(&self) -> Result<String> {
        Ok(self.executor("executor_id")?.executor_id().to_owned())
    }

    /// The version of the application's code, as workflow rows record it.
    pub fn app_version(&self) -> Result<String> {
        Ok(self.executor("app_version")?.app_version().to_owned())
    }

    /// This application's DBOS Cloud id, empty off DBOS Cloud.
    pub fn app_id(&self) -> Result<String> {
        Ok(self.executor("app_id")?.app_id().to_owned())
    }

    /// The running executor, or [`Error::NotLaunched`] naming the operation that wanted it.
    ///
    /// The operation is named because the useful half of the error is which call was too early.
    pub(crate) fn executor(&self, operation: &'static str) -> Result<Arc<Executor>> {
        self.0.executor.get(operation)
    }

    /// The slot this instance's executor is installed in, for a [`WorkflowRef`](crate::WorkflowRef)
    /// to hold in place of the instance.
    pub(crate) fn executor_slot(&self) -> Arc<ExecutorSlot> {
        Arc::clone(&self.0.executor)
    }

    fn read_executor(&self) -> Option<Arc<Executor>> {
        self.0.executor.read()
    }

    fn write_executor(&self) -> std::sync::RwLockWriteGuard<'_, Option<Arc<Executor>>> {
        self.0.executor.write()
    }
}

/// Where an instance's executor is installed: `None` until launched and again after shutdown.
///
/// **Its own allocation rather than a field of the instance**, so that a
/// [`WorkflowRef`](crate::WorkflowRef) can hold it without holding the instance. A workflow body
/// starts a child through a reference captured in its registered closure, and the registry that
/// owns the closure lives on the instance, so a reference that held the instance would hold
/// itself: the instance, its registry and its configuration would never be freed. Holding the
/// slot instead makes a cycle only through the executor — slot, executor, its registry snapshot,
/// the closure, the reference, the slot — and shutdown breaks it by emptying the slot. A
/// reference also starts workflows for as long as the instance is launched, whether or not any
/// handle to the instance itself is still held.
///
/// Read on every operation, written twice in a process's life, so a reader-biased lock rather than
/// an async one.
#[derive(Default)]
pub(crate) struct ExecutorSlot(RwLock<Option<Arc<Executor>>>);

impl ExecutorSlot {
    /// The running executor, or [`Error::NotLaunched`] naming the operation that wanted it.
    pub(crate) fn get(&self, operation: &'static str) -> Result<Arc<Executor>> {
        self.read().ok_or(Error::NotLaunched {
            operation: operation.into(),
        })
    }

    fn read(&self) -> Option<Arc<Executor>> {
        // A poisoned lock here cannot mean torn state: the guarded value is one `Option<Arc<_>>`
        // and every section under the lock is infallible, so a panic elsewhere leaves it whole.
        self.0
            .read()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clone()
    }

    fn write(&self) -> std::sync::RwLockWriteGuard<'_, Option<Arc<Executor>>> {
        self.0
            .write()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }
}

impl std::fmt::Debug for DBOS {
    /// Hand-written because the derive would print the whole configuration, including the database
    /// URL and whatever credentials it carries.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("DBOS")
            .field("app_name", &self.0.config.app_name)
            .field("launched", &self.is_launched())
            .finish_non_exhaustive()
    }
}

/// Registers this version and returns what a previous run of this executor abandoned to its queue.
///
/// One function rather than two calls inline so that [`Executor::start`] has a single error path
/// to close the system database on; see the comment at its call site. Everything between the
/// connect and the executor being built belongs here for that reason, and anything added later
/// belongs here too.
async fn prepare(
    sysdb: &dyn SystemDatabase,
    executor_id: &str,
    app_version: &str,
) -> Result<Vec<String>> {
    register_version(sysdb, app_version).await?;
    crate::recovery::reenqueue(sysdb, executor_id, app_version).await
}

/// Registers this version and warns if it is not the one a rolling deploy would prefer.
///
/// A warning rather than a refusal: running an older version alongside a newer one is exactly what
/// a rollback looks like, and it is the deployment's call. What it must not be is silent, because
/// the symptom — an executor that starts cleanly and is handed no unversioned work — reads as a
/// broken queue rather than as a deliberate policy.
async fn register_version(sysdb: &dyn SystemDatabase, version: &str) -> Result<()> {
    sysdb
        .create_application_version(version, None)
        .await
        .map_err(Error::SystemDatabase)?;
    match sysdb
        .get_latest_application_version(None)
        .await
        .map_err(Error::SystemDatabase)?
    {
        Some(latest) if latest.version_name != version => {
            tracing::warn!(
                app_version = version,
                latest_version = latest.version_name,
                "this executor is not running the latest registered application version: it will \
                 recover and dequeue only work stamped with its own version"
            );
        }
        _ => {}
    }
    Ok(())
}

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

    fn config() -> Config {
        Config {
            app_version: Some("1.0.0".to_owned()),
            ..Config::new("test-app", "postgres://localhost/nothing")
        }
    }

    #[tokio::test]
    async fn nothing_is_launched_until_launch() {
        let dbos = DBOS::new(config());
        assert!(!dbos.is_launched());
        let err = dbos.app_version().unwrap_err();
        assert!(
            matches!(
                err,
                Error::NotLaunched {
                    operation: std::borrow::Cow::Borrowed("app_version")
                }
            ),
            "{err}"
        );
    }

    #[tokio::test]
    async fn not_launched_names_the_operation_that_was_too_early() {
        let dbos = DBOS::new(config());
        let err = dbos.executor("do_something").unwrap_err();
        assert_eq!(
            err.to_string(),
            "cannot do_something before DBOS is launched"
        );
    }

    #[test]
    fn the_lifecycle_futures_are_send() {
        // A `std::sync::RwLockWriteGuard` is `!Send`, so a future holding one across an await is
        // `!Send` and this stops compiling. That makes the "no `std` guard crosses an await" rule
        // a compile-time check rather than a comment — and `shutdown` is where it is easy to get
        // wrong, since it takes the write guard and then awaits the executor's teardown.
        fn assert_send<F: Send>(_: F) {}
        let dbos = DBOS::new(config());
        assert_send(dbos.launch());
        assert_send(dbos.shutdown());
    }

    #[tokio::test]
    async fn shutting_down_an_unlaunched_instance_does_nothing() {
        let dbos = DBOS::new(config());
        dbos.shutdown().await;
        assert!(!dbos.is_launched());
    }

    #[tokio::test]
    async fn an_invalid_config_is_refused_before_anything_is_connected() {
        let dbos = DBOS::new(Config {
            app_version: Some("1.0.0".to_owned()),
            ..Config::new("No", "postgres://localhost/nothing")
        });
        let err = dbos.launch().await.unwrap_err();
        assert!(matches!(err, Error::Config(_)), "{err}");
        assert!(!dbos.is_launched());
    }

    /// The registry lives on the instance, so anything a registered closure captures is held by
    /// the instance holding it. A closure capturing a `DBOS` is therefore a cycle and the
    /// instance — with its executor and its connection pool — is never freed, which is why
    /// nothing the engine puts in there may capture one. This pins that: `register_workflow` must
    /// not quietly store a handle to `self` alongside the caller's closure, and the workflow-facing
    /// calls must stay reachable without one.
    #[test]
    fn registration_does_not_make_the_instance_hold_itself() {
        let dbos = DBOS::new(config());
        let alive = Arc::downgrade(&dbos.0);
        dbos.register_workflow("noop", |()| async { Ok::<_, Error>(()) })
            .unwrap();
        drop(dbos);
        assert_eq!(
            alive.strong_count(),
            0,
            "the instance outlived every handle to it: something in the registry holds it"
        );
    }

    #[test]
    fn debug_does_not_print_the_database_url() {
        let dbos = DBOS::new(Config::new("test-app", "postgres://user:hunter2@host/db"));
        let shown = format!("{dbos:?}");
        assert!(!shown.contains("hunter2"), "{shown}");
        assert!(shown.contains("test-app"), "{shown}");
    }

    /// A launched instance against `db`, under its own application name and version.
    fn launchable(app_name: &str, db: &dbos_test_support::TestDatabase) -> DBOS {
        DBOS::new(Config {
            migrate: false,
            app_version: Some(format!("{app_name}-1.0.0")),
            ..Config::new(app_name, db.url())
        })
    }

    /// Shuts down the executor in `slot` the way [`DBOS::shutdown`] would, for a test that has
    /// dropped every handle to the instance and so has nothing to call it on.
    async fn shut_down(slot: &ExecutorSlot) {
        let executor = slot.write().take().expect("still launched");
        executor.stop(Duration::ZERO).await;
        executor.close().await;
    }

    /// A `WorkflowRef` keeps starting workflows while its instance is launched, after the last
    /// handle to the instance itself is dropped: it holds the executor's slot, not the instance.
    ///
    /// A unit test because nothing outside the crate can shut such an instance down.
    #[tokio::test]
    async fn a_workflow_ref_starts_workflows_after_the_instance_is_dropped() {
        let db = dbos_test_support::test_database().await;
        let dbos = launchable("surviving-ref", &db);
        let child = dbos
            .register_workflow("child", |n: u32| async move { Ok::<u32, Error>(n * 2) })
            .expect("registration failed");
        let parent = {
            let child = child.clone();
            dbos.register_workflow("parent", move |n: u32| {
                let child = child.clone();
                async move { child.run(n).await }
            })
            .expect("registration failed")
        };
        dbos.launch().await.expect("launch failed");
        let slot = dbos.executor_slot();
        drop(dbos);

        assert_eq!(child.run(3).await.expect("the run failed"), 6);
        let handle = child.start(4).await.expect("the start failed");
        assert_eq!(handle.result().await.expect("the child failed"), 8);
        assert_eq!(
            parent.run(5).await.expect("the parent failed"),
            10,
            "a child started from a workflow body, with no handle to the instance anywhere"
        );

        shut_down(&slot).await;
    }

    /// A child started through a reference from another instance is still
    /// [`Error::WrongInstance`] once every handle to that other instance is dropped, rather than
    /// some other error that hides the wiring mistake.
    #[tokio::test]
    async fn a_dropped_instance_s_workflow_ref_is_still_the_wrong_instance() {
        let db = dbos_test_support::test_database().await;
        let other = launchable("dropped-other-instance", &db);
        let owner = launchable("dropped-owner-instance", &db);
        let child = other
            .register_workflow("child", |()| async move { Ok::<u32, Error>(1) })
            .expect("registration failed");
        let parent = owner
            .register_workflow("parent", move |()| {
                let child = child.clone();
                async move { child.run(()).await }
            })
            .expect("registration failed");
        other.launch().await.expect("launch failed");
        owner.launch().await.expect("launch failed");
        let other_slot = other.executor_slot();
        drop(other);

        let error = parent.run(()).await.expect_err("the parent succeeded");
        assert!(
            matches!(&error, Error::WrongInstance { operation } if operation.contains("workflow")),
            "expected a wrong-instance refusal, got {error:?}"
        );

        owner.shutdown().await;
        shut_down(&other_slot).await;
    }
}