dbos 0.6.0

//! Debouncing: many requests within a period collapse into one workflow run.
//!
//! Debouncing a key starts a workflow that waits `DELAYED` on a queue, holding a deduplication id
//! derived from the key. Each later debounce of the key pushes that workflow's release out by the
//! period it names and replaces its input, so it runs **once**, with the **last** input, when the
//! calls stop — or at the debounce timeout, which caps how far a steady stream of calls can push
//! it. Once the workflow is released to its queue the key is free again, and the next debounce
//! starts a fresh workflow.
//!
//! [`Debouncer`] debounces a registered workflow from an app; [`DebouncerClient`] does the same
//! from a [`Client`], naming the workflow as an enqueue does.
//!
//! # The loop
//!
//! A debounce is a loop around two writes, because nothing makes them one:
//!
//! 1. **Bounce.** [`debounce_delayed_workflow`](crate::sysdb::SystemDatabase::debounce_delayed_workflow)
//!    extends the debounced workflow holding the key, if there is one this debounce may extend, and
//!    the caller gets a handle to it.
//! 2. **Otherwise, look at who holds the key.** Nobody: go on to 3. A workflow that is not
//!    debounced, is a different workflow (name, class or configured instance), or belongs to
//!    another app: the key is not this debounce's, and the call fails with the deduplication error
//!    an enqueue under a held key fails with. A debounced holder of the same workflow that the
//!    bounce could not extend is between states; pause briefly and bounce again.
//! 3. **Enqueue** a fresh debounced workflow under the key. A concurrent debounce may enqueue
//!    first, and then this insert fails on the deduplication index: go back to 1, which now
//!    extends the winner.

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

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

use crate::checkpoint::{PendingStep, StepPlacement};
use crate::client::Client;
use crate::enqueue::EnqueueOptions;
use crate::error::{DurableError, EngineOnly, Error, Result};
use crate::handle::WorkflowHandle;
use crate::identity::refuse_app_name;
use crate::instance::Executor;
use crate::registry::{QualifiedWorkflowName, WorkflowRef};
use crate::serialization::{PayloadEncoding, SerializationFormat, encode_input};
use crate::sysdb::INTERNAL_QUEUE;
use crate::sysdb::types::{
    Debounce, DebounceRequest, Timestamp, WorkflowIdReusePolicy, step_names,
    validate_positive_duration,
};
use crate::workflow::{
    Debounced, OwnedQueueRequest, OwnedStart, PendingStart, QueueRequest, Role, TargetVersion,
    Timeout, create, encode_attributes, placed_workflow_id, refused_start, spawn_tracked,
    validate_start,
};

/// What a debounce refuses its arguments as.
const DEBOUNCE_OPERATION: &str = "debounce a workflow";

/// How long a debounce waits before bouncing again when the key's debounced holder is between
/// states, so a holder stuck there does not have the loop spinning against the database.
const RETRY_PAUSE: Duration = Duration::from_millis(100);

/// The deduplication id a debounce of `key` holds for `workflow`: `<workflow>-<key>`, with the
/// workflow spelled `name`, `name/class`, `name/class/config`, or `name//config` for a config with
/// no class.
///
/// The whole name, so two configured instances of one class, or a method and a free function of
/// the same bare name, debounce one key independently rather than colliding on it. The parts are
/// joined without escaping, so names that contain `/` or `-` can still spell the same id — a free
/// function `a/b` and a method `a` of class `b`, or `a` with key `b-c` and `a-b` with key `c`.
/// Such a collision is refused rather than extended: [`classify`] compares the holder's whole
/// name.
///
/// **The one place the spelling is written**, so changing it is a change here and nowhere else.
/// It is stored, and only a debounce that spells the id this way finds a workflow another debounce
/// of the key started, so it is spelled from the key's parts rather than from its
/// [`Display`](std::fmt::Display), which is for people to read.
pub(crate) fn deduplication_id(workflow: &QualifiedWorkflowName, key: &str) -> String {
    let name = workflow.name();
    match (workflow.class_name(), workflow.config_name()) {
        (None, None) => format!("{name}-{key}"),
        (Some(class), None) => format!("{name}/{class}-{key}"),
        (Some(class), Some(config)) => format!("{name}/{class}/{config}-{key}"),
        (None, Some(config)) => format!("{name}//{config}-{key}"),
    }
}

/// What one debounce says beyond its key, period and input — see
/// [`Debouncer::debounce_with`] and [`DebouncerClient::debounce_with`].
///
/// Per call rather than on the debouncer, because what it names belongs to one workflow: a
/// debouncer serves any number of keys, and any number of workflows one key after another.
#[derive(Debug, Clone, Copy, Default)]
pub struct DebounceOptions<'a> {
    /// The id of the workflow this call enqueues, in place of a generated one.
    ///
    /// Used **only when the call enqueues a fresh workflow**. A call that coalesces onto a
    /// workflow already waiting under the key hands back that workflow, whatever id this names.
    /// A fresh enqueue under an id that already names a workflow is not joined to it: the call
    /// fails with [`ConflictingWorkflow`](crate::sysdb::Error::ConflictingWorkflow).
    pub workflow_id: Option<&'a str>,
}

/// What a debounce call does about one bounce's outcome.
#[derive(Debug, PartialEq, Eq)]
enum BounceAction {
    /// The bounce extended this workflow, already waiting under the key; the call coalesces onto
    /// it and hands back a handle to it.
    Coalesce(String),
    /// Nothing holds the key; enqueue a fresh workflow.
    Enqueue,
    /// The key is held by this workflow, which this debounce must not touch; the call fails as a
    /// deduplicated enqueue would.
    Refuse(String),
    /// A debounced holder of this workflow could not be extended this time; bounce again.
    Retry,
}

/// Decides what a debounce of `workflow`, acting for `target_app`, does about `outcome`.
///
/// A holder that is not debounced is an ordinary deduplicated enqueue, and one of a different
/// name, class or configured instance is a key collision between unrelated workflows (`"a"` with
/// key `"b-c"` against `"a-b"` with `"c"`): both are refused rather than hijacked. So is a holder
/// another app owns, which the bounce never extends — retrying it would spin until the holder
/// left. A holder with no owner, or a debounce acting for no app, cannot be told apart that way,
/// and is retried.
fn classify(
    outcome: Debounce,
    workflow: &QualifiedWorkflowName,
    target_app: Option<&str>,
) -> BounceAction {
    match outcome {
        Debounce::Bounced { workflow_id } => BounceAction::Coalesce(workflow_id),
        Debounce::Unheld => BounceAction::Enqueue,
        Debounce::Held(holder) => {
            // Part by part, with `''` absent as `QualifiedWorkflowName::from_row` reads it.
            fn present(part: Option<&str>) -> Option<&str> {
                part.filter(|part| !part.is_empty())
            }
            let same_workflow = holder.workflow_name.as_deref() == Some(workflow.name())
                && present(holder.class_name.as_deref()) == workflow.class_name()
                && present(holder.config_name.as_deref()) == workflow.config_name();
            let another_app = matches!(
                (target_app, holder.application_name.as_deref()),
                (Some(target), Some(owner)) if target != owner
            );
            if !holder.is_debounced || !same_workflow || another_app {
                BounceAction::Refuse(holder.workflow_id)
            } else {
                BounceAction::Retry
            }
        }
    }
}

/// Everything a debounce's bounces share, settled once before the loop: what is debounced, under
/// which key and on which queue, with which input, for which app.
///
/// Both debouncers drive their loops through this, so a bounce is asked for and answered the same
/// way from an app and from a client.
struct Bounce<'a> {
    workflow: &'a QualifiedWorkflowName,
    queue_name: &'a str,
    deduplication_id: String,
    period: Duration,
    /// The input, encoded once: every bounce and the fresh enqueue write the same text.
    input: String,
    encoding: PayloadEncoding,
    target_app: Option<&'a str>,
}

impl Bounce<'_> {
    /// The request for one bounce, releasing the workflow `period` from it.
    ///
    /// The period goes as a duration, which the bounce resolves on the database clock, so each
    /// attempt measures it from when that attempt lands.
    fn request(&self) -> DebounceRequest<'_> {
        DebounceRequest {
            workflow: self.workflow,
            queue_name: self.queue_name,
            deduplication_id: &self.deduplication_id,
            period: self.period,
            inputs: Some(&self.input),
            serialization: Some(self.encoding.name()),
            application_name: self.target_app,
        }
    }

    /// What to do about a bounce's `outcome` — see [`classify`].
    fn action(&self, outcome: Debounce) -> BounceAction {
        classify(outcome, self.workflow, self.target_app)
    }

    /// The error for a key `holder` holds that this debounce must not extend: the one an enqueue
    /// under a held key fails with.
    fn deduplicated(self, holder: String) -> Error {
        Error::SystemDatabase(crate::sysdb::Error::QueueDeduplicated {
            workflow_id: holder,
            queue_name: self.queue_name.to_owned(),
            deduplication_id: self.deduplication_id,
        })
    }
}

/// Whether a fresh enqueue lost the key to a concurrent debounce, which the loop answers by
/// bouncing again — the bounce then extends the winner.
fn lost_the_key(result: &Result<impl Sized>) -> bool {
    matches!(
        result,
        Err(Error::SystemDatabase(
            crate::sysdb::Error::QueueDeduplicated { .. }
        ))
    )
}

/// Refuses a period or a debounce timeout of zero, or one too long to store.
///
/// A zero period would enqueue the workflow ready to run rather than `DELAYED`, where no later
/// bounce can find it.
fn refuse_durations(period: Duration, debounce_timeout: Option<Duration>) -> Result<()> {
    validate_positive_duration("period", period)
        .map_err(|error| Error::refused_argument(DEBOUNCE_OPERATION, error))?;
    if let Some(timeout) = debounce_timeout {
        validate_positive_duration("debounce_timeout", timeout)
            .map_err(|error| Error::refused_argument(DEBOUNCE_OPERATION, error))?;
    }
    Ok(())
}

/// A registered workflow, debounced.
///
/// Built with [`Debouncer::new`] and adjusted by functional update, and reusable: one debouncer
/// serves any number of keys.
///
/// ```no_run
/// # async fn f(workflow: &dbos::WorkflowRef<String, ()>) -> dbos::Result<()> {
/// use std::time::Duration;
///
/// let debouncer = dbos::Debouncer {
///     debounce_timeout: Some(Duration::from_secs(60)),
///     ..dbos::Debouncer::new(workflow)
/// };
/// // Runs once, with "third", two seconds after the last call — or a minute after the first.
/// for input in ["first", "second", "third"] {
///     debouncer.debounce("user-42", Duration::from_secs(2), input.to_owned()).await?;
/// }
/// # Ok(()) }
/// ```
///
/// # What a debouncer sets
///
/// A debouncer's fields apply to every workflow a debounce enqueues. They are the part of a start
/// a debounce leaves to its caller: the debounce writes the workflow's deduplication id (the key)
/// and its delay (the period) itself, joins the workflow holding its key rather than refusing or
/// returning it, and enqueues many workflows, so it takes no workflow id but the one a single call
/// names with [`debounce_with`](Self::debounce_with). Debounced workflows support neither a
/// priority nor a partition key.
///
/// # Inside a workflow
///
/// A debounce called from a workflow body is durable. Each bounce commits with its own step
/// checkpoint, and a fresh enqueue is recorded as a child start, so a replay reads back what the
/// first run did — the same handle, without bouncing or enqueueing again. Called inside a step, it
/// is [`Error::InsideStep`], as a start is.
///
/// **A fresh enqueue makes the debounced workflow the calling workflow's child**: its row names
/// the caller as its parent, as any child start's does. So cancelling the caller cancels the
/// debounced workflow too — and that workflow is shared: later debounces of the key, from other
/// callers, coalesce onto it, and the input it would have run with may be theirs. A debounce that
/// only extends a waiting workflow records no parent on it.
pub struct Debouncer<'a, P, R, E = EngineOnly> {
    /// The workflow to debounce.
    pub workflow: WorkflowRef<P, R, E>,
    /// The queue the debounced workflow waits and runs on, and the scope of its key: the same key
    /// on two queues is two debounces. `None` is the internal queue every app dequeues.
    pub queue: Option<&'a str>,
    /// How the workflow's input, output and error are encoded: the registration's for
    /// [`SerializationFormat::Default`], as a start's are.
    pub serialization: SerializationFormat,
    /// The workflow's budget once it runs, or `None` for none.
    ///
    /// **Never a calling workflow's deadline**: a debounce can hold the workflow back for longer
    /// than the caller lives.
    pub timeout: Option<Duration>,
    /// Attributes for the row of a fresh workflow.
    pub attributes: Option<&'a serde_json::Map<String, serde_json::Value>>,
    /// The longest a workflow may be held back, counted from the debounce that enqueued it, on the
    /// database clock.
    ///
    /// Later debounces extend the delay up to this cap and no further, so a steady stream of calls
    /// cannot postpone the workflow forever. `None` is no cap.
    pub debounce_timeout: Option<Duration>,
    /// The app the debounce acts for, or `None` for this one.
    ///
    /// A fresh workflow is written for that app — with no app version, so that app's latest runs
    /// it. A waiting workflow is extended if it is that app's or no app's, and one that was no
    /// app's becomes that app's.
    pub app_name: Option<&'a str>,
}

impl<'a, P, R, E> Debouncer<'a, P, R, E> {
    /// Debounces `workflow` on the internal queue, with no timeout, for this app.
    pub fn new(workflow: &WorkflowRef<P, R, E>) -> Self {
        Self {
            workflow: workflow.clone(),
            queue: None,
            serialization: SerializationFormat::Default,
            timeout: None,
            attributes: None,
            debounce_timeout: None,
            app_name: None,
        }
    }

    /// The queue a debounce enqueues onto, as an enqueue names it.
    fn queue_request(&self) -> QueueRequest<'a> {
        QueueRequest {
            serialization: self.serialization,
            ..QueueRequest::new(self.queue.unwrap_or(INTERNAL_QUEUE))
        }
    }
}

/// Hand-written: the derive would demand `P: Clone, R: Clone`, which a debouncer that holds
/// neither has no business requiring.
impl<P, R, E> Clone for Debouncer<'_, P, R, E> {
    fn clone(&self) -> Self {
        Self {
            workflow: self.workflow.clone(),
            queue: self.queue,
            serialization: self.serialization,
            timeout: self.timeout,
            attributes: self.attributes,
            debounce_timeout: self.debounce_timeout,
            app_name: self.app_name,
        }
    }
}

impl<P, R, E> std::fmt::Debug for Debouncer<'_, P, R, E> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Debouncer")
            .field("workflow", &self.workflow)
            .field("queue", &self.queue)
            .field("serialization", &self.serialization)
            .field("timeout", &self.timeout)
            .field("attributes", &self.attributes)
            .field("debounce_timeout", &self.debounce_timeout)
            .field("app_name", &self.app_name)
            .finish()
    }
}

impl<P, R, E> Debouncer<'_, P, R, E>
where
    P: Serialize + DeserializeOwned + Send + 'static,
    R: Serialize + DeserializeOwned + Send + 'static,
    E: DurableError,
{
    /// Debounces `key`: the workflow runs once, with the last `input`, `period` after the last
    /// debounce of the key — or at the [`debounce_timeout`](Self::debounce_timeout).
    ///
    /// Hands back a handle to the workflow this call fed: the one already waiting under the key,
    /// whose release it pushed out and whose input it replaced, or a fresh one. Every call that
    /// coalesces gets a handle to the same workflow, and its result is the run with the last
    /// input.
    ///
    /// # Errors
    ///
    /// - [`Error::InvalidArgument`] for a zero `period`, [`timeout`](Self::timeout) or
    ///   [`debounce_timeout`](Self::debounce_timeout), an empty [`queue`](Self::queue), or an
    ///   invalid [`app_name`](Self::app_name).
    /// - [`Error::SystemDatabase`] with
    ///   [`QueueDeduplicated`](crate::sysdb::Error::QueueDeduplicated) when the key is held by a
    ///   workflow this debounce must not extend — an ordinary deduplicated enqueue, another
    ///   workflow whose key spells the same, or another app's — naming the holder.
    /// - [`Error::InsideStep`] when called inside a step, and [`Error::WrongInstance`] when called
    ///   from a workflow another instance runs, as a start is.
    ///
    /// # Step ids
    ///
    /// Inside a workflow, a debounce claims two step ids where it is written: the bounce's, and the
    /// fresh enqueue's, which a debounce that extends a waiting workflow leaves unused. When an
    /// attempt neither extends nor enqueues — another debounce of the key enqueued between the
    /// two, or the bounce met a waiting workflow between states — this one claims two more for
    /// each further attempt as it makes it. So a debounce should be driven to completion before
    /// the workflow's next durable call, rather than joined with one.
    pub fn debounce<'b>(&'b self, key: &str, period: Duration, input: P) -> PendingStart<'b, R, E> {
        self.debounce_with(key, period, input, DebounceOptions::default())
    }

    /// [`debounce`](Self::debounce), with something to say about this one call.
    ///
    /// [`options.workflow_id`](DebounceOptions::workflow_id) names the workflow this call enqueues
    /// if it enqueues a fresh one. An id that already names a workflow is
    /// [`Error::SystemDatabase`] with
    /// [`ConflictingWorkflow`](crate::sysdb::Error::ConflictingWorkflow), never a join: a debounce
    /// that joined a finished workflow would drop its input.
    ///
    /// ```no_run
    /// # async fn f(debouncer: &dbos::Debouncer<'_, String, ()>) -> dbos::Result<()> {
    /// use std::time::Duration;
    ///
    /// let handle = debouncer
    ///     .debounce_with(
    ///         "user-42",
    ///         Duration::from_secs(2),
    ///         "latest".to_owned(),
    ///         dbos::DebounceOptions {
    ///             workflow_id: Some("user-42-digest-7"),
    ///         },
    ///     )
    ///     .await?;
    /// # Ok(()) }
    /// ```
    pub fn debounce_with<'b>(
        &'b self,
        key: &str,
        period: Duration,
        input: P,
        options: DebounceOptions<'b>,
    ) -> PendingStart<'b, R, E> {
        let deduplication_id = deduplication_id(self.workflow.qualified_name(), key);
        let built = self
            .place(period)
            .map(|(executor, bounce, start)| ((executor, start), bounce));
        PendingStart::from_step(PendingStep::placed(
            step_names::DEBOUNCE,
            built,
            move |(executor, start), bounce| async move {
                self.debounced(
                    executor,
                    bounce,
                    start,
                    deduplication_id,
                    period,
                    input,
                    options,
                )
                .await
            },
        ))
    }

    /// Everything a debounce settles where it is written: the instance, that its arguments are
    /// ones it can act on, and the two step ids it starts with.
    ///
    /// The checks come first, so a refused debounce moves no step counter.
    fn place(&self, period: Duration) -> Result<(Arc<Executor>, StepPlacement, StepPlacement)> {
        let executor = self.workflow.executor(DEBOUNCE_OPERATION)?;
        // A debounce names no workflow id of its own here: one call's is `DebounceOptions`'.
        validate_start(
            Some(&self.queue_request()),
            self.timeout,
            None,
            WorkflowIdReusePolicy::ReturnExisting,
        )?;
        refuse_durations(period, self.debounce_timeout)?;
        refuse_app_name(DEBOUNCE_OPERATION, self.app_name)?;
        let bounce = StepPlacement::of(executor.connection(), DEBOUNCE_OPERATION)?;
        if matches!(bounce, StepPlacement::InsideStep { .. }) {
            return Err(Error::InsideStep {
                operation: "debouncing a workflow".into(),
            });
        }
        let start = StepPlacement::of(executor.connection(), DEBOUNCE_OPERATION)?;
        Ok((executor, bounce, start))
    }

    /// The loop, once something polls the debounce. See the [module documentation](self).
    #[allow(clippy::too_many_arguments)]
    async fn debounced(
        &self,
        executor: Arc<Executor>,
        mut bounce_at: StepPlacement,
        mut start: StepPlacement,
        deduplication_id: String,
        period: Duration,
        input: P,
        call: DebounceOptions<'_>,
    ) -> Result<WorkflowHandle<R, E>> {
        let conn = executor.connection();
        let name: &QualifiedWorkflowName = self.workflow.qualified_name();
        let queue = self.queue_request();
        // The registration's encoding unless the queue asks for its own, as a start's is. A bounce
        // writes the input in the same one, so the row's tag always describes its input.
        let encoding = PayloadEncoding::resolve(
            queue.serialization,
            PayloadEncoding::resolve(
                self.workflow.serialization(),
                executor.serializer().encoding(),
            ),
        );
        let bounce = Bounce {
            workflow: name,
            queue_name: queue.name,
            deduplication_id,
            period,
            input: encode_input(&input, &encoding)?,
            encoding,
            target_app: Some(self.app_name.unwrap_or(executor.app_name())),
        };
        let attributes = encode_attributes(self.attributes)?;
        // Never the caller's deadline: the debounce can hold the workflow back for longer than the
        // caller has left.
        let timeout = self.timeout.map_or(Timeout::None, Timeout::Explicit);

        loop {
            let outcome = conn
                .sysdb()
                .debounce_delayed_workflow(
                    &bounce.request(),
                    conn.step_caller(bounce_at.step_writer()),
                )
                .await
                .map_err(|error| Error::refused_argument(DEBOUNCE_OPERATION, error))?;

            match bounce.action(outcome) {
                // The bounce matched the row, so a later absence is a deletion.
                BounceAction::Coalesce(workflow_id) => {
                    return Ok(WorkflowHandle::polling(Arc::clone(conn), workflow_id, true));
                }
                BounceAction::Refuse(holder) => return Err(bounce.deduplicated(holder)),
                BounceAction::Retry => crate::rt::sleep(RETRY_PAUSE).await,
                BounceAction::Enqueue => {
                    // The period is the delay and the timeout the deadline, both resolved by the
                    // insert on one reading of the database clock, which caps the release at the
                    // deadline. A replay that re-runs the start after a crash resolves them again,
                    // from when that insert lands.
                    let fresh = QueueRequest {
                        deduplication_id: Some(&bounce.deduplication_id),
                        delay: Some(period),
                        ..queue.clone()
                    };
                    let request = OwnedStart {
                        workflow_id: placed_workflow_id(call.workflow_id, &start),
                        // A debounced insert refuses an existing workflow under its id itself.
                        workflow_id_reuse_policy: WorkflowIdReusePolicy::ReturnExisting,
                        timeout,
                        queue: Some(OwnedQueueRequest::from(&fresh)),
                        input: Some(bounce.input.clone()),
                        encoding: bounce.encoding.clone(),
                        attributes: attributes.clone(),
                        started_at: Timestamp::now(),
                        debounce: Some(Debounced {
                            timeout: self.debounce_timeout,
                        }),
                        app_name: self.app_name.map(str::to_owned),
                        app_version: TargetVersion::Executor,
                    };
                    // Recorded at the start's step as a child start would be — the workflow it
                    // created, or the deduplication error it lost the race with — so a replay
                    // takes the same branch. The replayed form of that error is the same variant,
                    // read back from the start's record.
                    let created = self
                        .created(&executor, start, request)
                        .await
                        .map_err(|error| refused_start(Some(queue.name), error));
                    if !lost_the_key(&created) {
                        return created;
                    }
                }
            }

            // A further attempt, which takes ids of its own where it runs.
            bounce_at = StepPlacement::of(conn, DEBOUNCE_OPERATION)?;
            start = StepPlacement::of(conn, DEBOUNCE_OPERATION)?;
        }
    }

    /// One attempt's fresh enqueue, run to the end whatever becomes of the debounce awaiting it.
    ///
    /// **Detached, as a start's writing half is** (see `WorkflowRef::started` and
    /// [`DBOS::enqueue_with`](crate::DBOS::enqueue_with)), so a body that stops listening — a
    /// `select!` whose other branch won, a future dropped some other way — takes the
    /// [`JoinHandle`](crate::rt::JoinHandle) and leaves the work. A created workflow commits
    /// together with its record at the start's step; a refused one is two writes, the failed insert
    /// and then the refusal recorded at the step, and detached, the second always follows the first.
    /// The task also holds the parent's starts token for as long as it exists, so the parent waits
    /// for the enqueue before recording its own outcome; see `Ctx::starts`.
    ///
    /// **Dropping the debounce is not cancelling its parent.** A parent cancelled, or past its
    /// deadline, gives up its owner token, and `execute` waits for its starts only within the
    /// deadline: an enqueue that has not yet checked the parent's ownership is then refused, and no
    /// workflow is created, as for any start of a parent that no longer runs.
    ///
    /// **Shutdown is not one of the drops it outlives.** The task is a workflow's, so shutdown
    /// aborts it with the parent's, and the debounce reports [`Interrupted`](Error::Interrupted):
    /// either no row, or one recorded at the start's step that a replay finds.
    ///
    /// **Nor does detaching make a dropped debounce replay safely.** If the attempt lost the key, its
    /// refusal is recorded, and a replay reads it and goes round the loop again, taking ids the
    /// original run never took for it — ids its later calls took instead. A debounce is driven to
    /// completion, as the module says. Answers in the engine's channel, as `create` does.
    async fn created(
        &self,
        executor: &Arc<Executor>,
        start: StepPlacement,
        request: OwnedStart,
    ) -> Result<WorkflowHandle<R, E>> {
        let workflow_id = request.workflow_id.clone();
        // Taken before the spawn and moved into the task, so one aborted before its first poll
        // still gives it back.
        let tracked = match &start {
            StepPlacement::Recorded { ctx, .. } => Some(ctx.starts().token()),
            _ => None,
        };
        let creating = create::<R, E>(
            Arc::clone(executor),
            start,
            self.workflow.qualified_name().clone(),
            request,
        )
        .instrument(tracing::Span::current());
        let creating = async move {
            let _tracked = tracked;
            creating.await
        };
        let label = format!("debounce {workflow_id}");
        match spawn_tracked(executor, Role::Workflow, label, creating).await {
            Ok(created) => created,
            Err(join) if join.is_cancelled() => Err(Error::Interrupted { workflow_id }),
            Err(join) => std::panic::resume_unwind(join.into_panic()),
        }
    }
}

/// A workflow debounced through a [`Client`], named as an enqueue names it.
///
/// What [`Debouncer`] is for an app, for a process that does not register the workflow: built
/// with [`DebouncerClient::new`] and adjusted by functional update.
///
/// ```no_run
/// # async fn f(client: &dbos::Client) -> dbos::Result<()> {
/// use std::time::Duration;
///
/// let debouncer = dbos::DebouncerClient {
///     debounce_timeout: Some(Duration::from_secs(60)),
///     ..dbos::DebouncerClient::new(client, "summarise", "work")
/// };
/// let handle: dbos::WorkflowHandle<String> = debouncer
///     .debounce("user-42", Duration::from_secs(2), "the latest text")
///     .await?;
/// # Ok(()) }
/// ```
///
/// [`workflow`](Self::workflow) is the workflow's name, as [`Client::enqueue`] takes it: a bare
/// name for a free function, or a [`QualifiedWorkflowName`] naming its class and configured
/// instance. The other fields apply to every workflow a debounce enqueues, as the same fields of
/// [`EnqueueOptions`] would to an enqueue, and leave out what a debounce sets itself — see
/// [`Debouncer`](Debouncer#what-a-debouncer-sets).
///
/// **Not durable from inside a workflow body**, as a client's enqueue is not: a client is a
/// connection, not an execution, so a replayed body debounces again.
#[derive(Debug, Clone)]
pub struct DebouncerClient<'a> {
    /// The client to debounce through.
    pub client: Client,
    /// The workflow, by the name it is registered under.
    pub workflow: QualifiedWorkflowName,
    /// The queue the debounced workflow waits and runs on, and the scope of its key — see
    /// [`Debouncer::queue`].
    pub queue: &'a str,
    /// How the workflow's input, output and error are encoded: this client's
    /// [`serializer`](crate::ClientConfig::serializer) for [`SerializationFormat::Default`], as an
    /// enqueue's are.
    pub serialization: SerializationFormat,
    /// The workflow's budget once it runs, or `None` for none.
    pub timeout: Option<Duration>,
    /// Attributes for the row of a fresh workflow.
    pub attributes: Option<&'a serde_json::Map<String, serde_json::Value>>,
    /// The version of the application's code that must run the workflow — see
    /// [`EnqueueOptions::app_version`].
    pub app_version: Option<&'a str>,
    /// The app the debounce acts for, or `None` for this client's — see
    /// [`Debouncer::app_name`].
    pub app_name: Option<&'a str>,
    /// The longest a workflow may be held back, counted from the debounce that enqueued it — see
    /// [`Debouncer::debounce_timeout`].
    pub debounce_timeout: Option<Duration>,
}

impl<'a> DebouncerClient<'a> {
    /// Debounces `workflow` on `queue`, with no timeout, for this client's app.
    pub fn new(
        client: &Client,
        workflow: impl Into<QualifiedWorkflowName>,
        queue: &'a str,
    ) -> Self {
        Self {
            client: client.clone(),
            workflow: workflow.into(),
            queue,
            serialization: SerializationFormat::Default,
            timeout: None,
            attributes: None,
            app_version: None,
            app_name: None,
            debounce_timeout: None,
        }
    }

    /// What a debounce enqueues with, as an enqueue names it.
    fn enqueue_options(&self) -> EnqueueOptions<'a> {
        EnqueueOptions {
            app_name: self.app_name,
            app_version: self.app_version,
            timeout: self.timeout.map_or(Timeout::None, Timeout::Explicit),
            attributes: self.attributes,
            serialization: self.serialization,
            ..EnqueueOptions::new(self.queue)
        }
    }

    /// Debounces `key` — see [`Debouncer::debounce`], whose errors this shares, apart from the
    /// two only a workflow body can meet.
    ///
    /// The input is encoded as [`Client::enqueue_with`] encodes it, and the type parameters are the
    /// caller's assertion, as there.
    pub async fn debounce<P, R, E>(
        &self,
        key: &str,
        period: Duration,
        input: P,
    ) -> Result<WorkflowHandle<R, E>>
    where
        P: Serialize,
    {
        self.debounce_with(key, period, input, DebounceOptions::default())
            .await
    }

    /// [`debounce`](Self::debounce), with something to say about this one call — see
    /// [`Debouncer::debounce_with`].
    pub async fn debounce_with<P, R, E>(
        &self,
        key: &str,
        period: Duration,
        input: P,
        call: DebounceOptions<'_>,
    ) -> Result<WorkflowHandle<R, E>>
    where
        P: Serialize,
    {
        // The app name first: the enqueue's own checks would refuse a bad one too, but as an
        // enqueue's rather than this debounce's.
        refuse_app_name(DEBOUNCE_OPERATION, self.app_name)?;
        let options = self.enqueue_options();
        options.validate(&self.workflow)?;
        refuse_durations(period, self.debounce_timeout)?;

        let conn = self.client.connection();
        let encoding = PayloadEncoding::resolve(self.serialization, conn.serializer().encoding());
        let bounce = Bounce {
            workflow: &self.workflow,
            queue_name: self.queue,
            deduplication_id: deduplication_id(&self.workflow, key),
            period,
            input: encode_input(&input, &encoding)?,
            encoding,
            target_app: self.app_name.or(conn.app_name()),
        };
        loop {
            let outcome = conn
                .sysdb()
                .debounce_delayed_workflow(&bounce.request(), None)
                .await
                .map_err(|error| Error::refused_argument(DEBOUNCE_OPERATION, error))?;

            match bounce.action(outcome) {
                BounceAction::Coalesce(workflow_id) => {
                    return Ok(WorkflowHandle::polling(Arc::clone(conn), workflow_id, true));
                }
                BounceAction::Refuse(holder) => return Err(bounce.deduplicated(holder)),
                BounceAction::Retry => crate::rt::sleep(RETRY_PAUSE).await,
                BounceAction::Enqueue => {
                    // The period is the delay and the timeout the deadline, both resolved by the
                    // insert on the database clock, which caps the release at the deadline.
                    let options = EnqueueOptions {
                        deduplication_id: Some(&bounce.deduplication_id),
                        delay: Some(period),
                        workflow_id: call.workflow_id,
                        app_name: bounce.target_app,
                        ..options.clone()
                    };
                    let created = self
                        .client
                        .enqueue_encoded(
                            &self.workflow,
                            &bounce.input,
                            bounce.encoding.clone(),
                            options,
                            Some(Debounced {
                                timeout: self.debounce_timeout,
                            }),
                        )
                        .await;
                    if !lost_the_key(&created) {
                        return created;
                    }
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sysdb::types::DebounceHolder;

    /// Each shape of key spells its own deduplication id, the stored form a later debounce of the
    /// key must reproduce.
    #[test]
    fn each_key_spells_its_deduplication_id() {
        let id = |workflow: QualifiedWorkflowName| deduplication_id(&workflow, "k");
        assert_eq!(id(QualifiedWorkflowName::new("f")), "f-k");
        assert_eq!(id(QualifiedWorkflowName::method("f", "C")), "f/C-k");
        assert_eq!(
            id(QualifiedWorkflowName::instance("f", "C", "eu")),
            "f/C/eu-k"
        );
        assert_eq!(
            id(QualifiedWorkflowName::instance("f", "", "eu")),
            "f//eu-k"
        );
    }

    fn holder(name: &str, is_debounced: bool, app: Option<&str>) -> Debounce {
        Debounce::Held(DebounceHolder {
            workflow_id: "holder".to_owned(),
            is_debounced,
            workflow_name: Some(name.to_owned()),
            class_name: None,
            config_name: None,
            application_name: app.map(str::to_owned),
        })
    }

    /// The spelling is `<workflow>-<key>`, which is what makes the two collide in
    /// `a_bounce_is_classified_by_its_holder`.
    #[test]
    fn the_deduplication_id_is_the_workflow_then_the_key() {
        let free = |name: &str| QualifiedWorkflowName::new(name);
        assert_eq!(deduplication_id(&free("wf"), "key"), "wf-key");
        assert_eq!(
            deduplication_id(&free("a"), "b-c"),
            deduplication_id(&free("a-b"), "c")
        );
    }

    /// The whole name is in the id, so a method, its configured instances and a free function of
    /// the same bare name each hold a key of their own.
    #[test]
    fn the_deduplication_id_spells_the_whole_name() {
        let eu = QualifiedWorkflowName::instance("wf", "Billing", "eu");
        let us = QualifiedWorkflowName::instance("wf", "Billing", "us");
        let method = QualifiedWorkflowName::method("wf", "Billing");
        assert_eq!(deduplication_id(&eu, "key"), "wf/Billing/eu-key");
        assert_eq!(deduplication_id(&method, "key"), "wf/Billing-key");
        let ids = [
            deduplication_id(&eu, "key"),
            deduplication_id(&us, "key"),
            deduplication_id(&method, "key"),
            deduplication_id(&QualifiedWorkflowName::new("wf"), "key"),
        ];
        let distinct: std::collections::HashSet<_> = ids.iter().collect();
        assert_eq!(distinct.len(), ids.len(), "{ids:?}");
    }

    #[test]
    fn a_bounce_is_classified_by_its_holder() {
        let wf = QualifiedWorkflowName::new("wf");
        assert_eq!(
            classify(
                Debounce::Bounced {
                    workflow_id: "w".to_owned()
                },
                &wf,
                Some("app")
            ),
            BounceAction::Coalesce("w".to_owned())
        );
        assert_eq!(
            classify(Debounce::Unheld, &wf, Some("app")),
            BounceAction::Enqueue
        );
        // An ordinary deduplicated enqueue, and a colliding key of another workflow.
        assert_eq!(
            classify(holder("wf", false, Some("app")), &wf, Some("app")),
            BounceAction::Refuse("holder".to_owned())
        );
        assert_eq!(
            classify(holder("wf-b", true, Some("app")), &wf, Some("app")),
            BounceAction::Refuse("holder".to_owned())
        );
        // Another configured instance of the same class and name.
        let east = QualifiedWorkflowName::instance("wf", "Billing", "east");
        let west = Debounce::Held(DebounceHolder {
            workflow_id: "holder".to_owned(),
            is_debounced: true,
            workflow_name: Some("wf".to_owned()),
            class_name: Some("Billing".to_owned()),
            config_name: Some("west".to_owned()),
            application_name: None,
        });
        assert_eq!(
            classify(west, &east, Some("app")),
            BounceAction::Refuse("holder".to_owned())
        );
        // A holder that spells an absent class or instance as `""` is the same workflow, which the
        // bounce extends too; one it could not extend this time is between states.
        let blank = Debounce::Held(DebounceHolder {
            workflow_id: "holder".to_owned(),
            is_debounced: true,
            workflow_name: Some("wf".to_owned()),
            class_name: Some(String::new()),
            config_name: Some(String::new()),
            application_name: Some("app".to_owned()),
        });
        assert_eq!(classify(blank, &wf, Some("app")), BounceAction::Retry);
        // Another application's holder is refused, never retried: the bounce cannot extend it.
        assert_eq!(
            classify(holder("wf", true, Some("other")), &wf, Some("app")),
            BounceAction::Refuse("holder".to_owned())
        );
        // Our own, or one no application can be told apart by, is between states.
        assert_eq!(
            classify(holder("wf", true, Some("app")), &wf, Some("app")),
            BounceAction::Retry
        );
        assert_eq!(
            classify(holder("wf", true, Some("other")), &wf, None),
            BounceAction::Retry
        );
        assert_eq!(
            classify(holder("wf", true, None), &wf, Some("app")),
            BounceAction::Retry
        );
    }

    #[test]
    fn a_zero_period_or_timeout_is_refused() {
        assert!(matches!(
            refuse_durations(Duration::ZERO, None),
            Err(Error::InvalidArgument { .. })
        ));
        assert!(matches!(
            refuse_durations(Duration::from_secs(1), Some(Duration::ZERO)),
            Err(Error::InvalidArgument { .. })
        ));
        refuse_durations(Duration::from_secs(1), Some(Duration::from_secs(1))).expect("valid");
    }

    /// An empty or malformed `app_name` is refused as `app_name`, not as a column.
    #[test]
    fn a_bad_app_name_is_refused() {
        for bad in ["", "No"] {
            let error = refuse_app_name(DEBOUNCE_OPERATION, Some(bad)).expect_err(bad);
            assert!(
                matches!(&error, Error::InvalidArgument { operation, detail }
                    if operation == DEBOUNCE_OPERATION
                        && detail.contains("`app_name`")
                        && !detail.contains("application_name")),
                "{error}"
            );
        }
        refuse_app_name(DEBOUNCE_OPERATION, Some("some-app")).expect("a valid name");
        refuse_app_name(DEBOUNCE_OPERATION, None).expect("no name");
    }
}