dbos 0.6.0

//! The shapes the system database stores and returns.
//!
//! Two conventions run through all of these, and both are deliberate.
//!
//! **Payloads are opaque.** Workflow inputs, outputs, and errors cross this boundary as
//! already-encoded strings. This layer never serializes or deserializes them, never inspects
//! them, and does not know which format they are in — the `serialization` column records that
//! for whoever reads them next. It is what lets one workflow's `rust_serde` payload and
//! another's `portable_json` sit in the same table, and what keeps this layer usable from a
//! language that does not share Rust's idea of a value.
//!
//! Besides two fixed values in a fixed encoding — a sleep's wake time and the stream-close
//! sentinel, both `portable_json` — the one value this layer puts into a column itself is the
//! result a management call records as its caller's step: a listing, a forked id, a recorded
//! refusal. Even that is encoded by the caller: it goes to the [`StepCaller`]'s [`StepCodec`] as
//! JSON data, and the row is tagged with whatever the codec names.
//!
//! **Times are typed, because the columns come in three shapes that a bare integer conflates.**
//! Some are instants ([`Timestamp`]); some are durations ([`std::time::Duration`]); one, in the
//! same table as both, is neither — `recovery_attempts` is a count. Worse, the durations do not
//! share a unit: `workflow_timeout_ms` is integer milliseconds while `polling_interval_sec` is a
//! float in seconds. `workflow_deadline_epoch_ms` and `workflow_timeout_ms` sit next to each
//! other, which is precisely the pair that gets swapped.
//!
//! [`Timestamp`] is a thin wrapper over the epoch milliseconds actually stored, so it converts
//! exactly rather than through someone's calendar, and `Duration` is in `std`. Callers wanting a
//! calendar type convert at their own edge.
//!
//! One exception, and it is the schema's rather than a preference: `workflow_schedules.last_fired_at`
//! holds ISO-8601 text, so a calendar is unavoidable for that column. `std` has none, so
//! [`Timestamp::to_iso8601`] and [`Timestamp::parse_iso8601`] go through `time` — parsing and
//! formatting only, with no timezone database, since the column is always UTC.

use super::Error;
use std::fmt;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use time::{OffsetDateTime, format_description::well_known::Rfc3339};

/// An instant, as epoch milliseconds.
///
/// Exactly what the columns hold, so reading and writing are lossless.
#[derive(
    Debug,
    Clone,
    Copy,
    Default,
    PartialEq,
    Eq,
    PartialOrd,
    Ord,
    Hash,
    serde::Serialize,
    serde::Deserialize,
)]
pub struct Timestamp(i64);

impl Timestamp {
    /// Wraps a stored value.
    pub const fn from_epoch_ms(ms: i64) -> Self {
        Self(ms)
    }

    /// The stored value.
    pub const fn as_epoch_ms(self) -> i64 {
        self.0
    }

    /// The current time, truncated to milliseconds.
    ///
    /// The `as` cast is safe by construction rather than by luck: it narrows a `u128` of
    /// milliseconds since 1970, which does not reach `i64::MAX` until the year 292{,}277{,}024.
    /// Elsewhere a caller supplies the duration and the conversion has to report — see
    /// [`checked_add`](Self::checked_add) — but here the clock supplies it.
    ///
    /// A clock set before 1970 gives the epoch rather than panicking. That is the lesser wrong:
    /// the timestamps this layer writes are for ordering and display, and a machine with a broken
    /// clock should record misleading times rather than fail every workflow it touches.
    pub fn now() -> Self {
        Self(
            SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .map(|d| d.as_millis() as i64)
                .unwrap_or(0),
        )
    }

    /// Converts to a `SystemTime`, or `None` for an instant before the epoch.
    ///
    /// No DBOS timestamp should be negative; this reports rather than saturates so a corrupt
    /// value surfaces instead of becoming 1970.
    pub fn to_system_time(self) -> Option<SystemTime> {
        u64::try_from(self.0)
            .ok()
            .map(|ms| UNIX_EPOCH + Duration::from_millis(ms))
    }

    /// Converts from a `SystemTime`, or `None` if it cannot be stored.
    ///
    /// Two ways it cannot: an instant before the epoch, and one so far after it that the
    /// milliseconds overflow `i64`. Both report rather than saturate, for the reason given on
    /// [`to_system_time`](Self::to_system_time) — a value this layer cannot hold is the caller's
    /// to hear about, not one to replace with a different instant.
    ///
    /// The second case is reachable where [`now`](Self::now)'s is not: the instant comes from the
    /// caller, and a `SystemTime` can hold seconds that do not survive being multiplied by a
    /// thousand.
    pub fn from_system_time(time: SystemTime) -> Option<Self> {
        time.duration_since(UNIX_EPOCH)
            .ok()
            .and_then(|d| i64::try_from(d.as_millis()).ok())
            .map(Self)
    }

    /// This instant plus a duration — a deadline from a start and a timeout.
    ///
    /// The reason the two are different types: adding a timeout to a deadline, or a deadline to
    /// a deadline, will not compile.
    pub fn checked_add(self, duration: Duration) -> Option<Self> {
        i64::try_from(duration.as_millis())
            .ok()
            .and_then(|ms| self.0.checked_add(ms))
            .map(Self)
    }

    /// This instant plus a duration the caller supplied, or [`Error::InvalidInput`] naming
    /// `field` when the sum cannot be stored.
    ///
    /// [`checked_add`](Self::checked_add) for a value that is about to be written: a duration too
    /// long for [`duration_to_ms`], or one that carries the instant past `i64::MAX` milliseconds,
    /// is the caller's to hear about rather than a wrapped instant in a column.
    pub fn try_add(self, field: &'static str, duration: Duration) -> Result<Self, Error> {
        let ms = duration_to_ms(field, duration)?;
        self.0
            .checked_add(ms)
            .map(Self)
            .ok_or_else(|| Error::InvalidInput {
                field: field.into(),
                detail: "must not end past the latest instant 64-bit epoch milliseconds can hold"
                    .to_owned(),
            })
    }

    /// How long after `earlier` this instant is, or `None` if it is not after it.
    pub fn duration_since(self, earlier: Self) -> Option<Duration> {
        self.0
            .checked_sub(earlier.0)
            .and_then(|ms| u64::try_from(ms).ok())
            .map(Duration::from_millis)
    }

    /// Reads an ISO-8601 instant, or `None` if the text is not one.
    ///
    /// The inverse of [`to_iso8601`](Self::to_iso8601), and it has to read whatever the other SDKs
    /// and Conductor sharing the database wrote: a `+00:00` offset, a `.000Z` fraction, a
    /// nanosecond fraction, a bare `Z` and an explicit numeric offset all parse to the same
    /// instant.
    ///
    /// Strict RFC 3339 otherwise: an offset is required and must carry its colon. That is
    /// stricter than a hand-written reader would be tempted to make it, and it turns away nothing
    /// any writer produces — Conductor's own reader draws the line in the same place.
    ///
    /// **Sub-millisecond precision is truncated**, since that is all [`Timestamp`] holds. A
    /// writer may store nanoseconds; cron granularity is seconds, so the digits being dropped
    /// cannot change a firing decision.
    pub fn parse_iso8601(text: &str) -> Option<Self> {
        let parsed = OffsetDateTime::parse(text, &Rfc3339).ok()?;
        // Nanoseconds to milliseconds, flooring, so an instant before the epoch truncates towards
        // the earlier millisecond rather than towards zero.
        i64::try_from(parsed.unix_timestamp_nanos().div_euclid(1_000_000))
            .ok()
            .map(Self)
    }

    /// Formats as ISO-8601 in UTC: `2026-08-12T14:30:00.123Z`, or `2026-08-12T14:30:00Z` on a
    /// whole second.
    ///
    /// For [`workflow_schedules.last_fired_at`](crate::sysdb::SystemDatabase::update_schedule_last_fired_at),
    /// the one column in the schema holding a formatted instant rather than epoch milliseconds.
    /// RFC 3339: a bare `Z`, with a fraction only when there is one and its trailing zeros
    /// trimmed, so 120 ms is `.12Z`.
    ///
    /// Infallible in practice and `String` rather than `Result` because of it: the only way
    /// `time` refuses to format is a year outside its range, which is four orders of magnitude
    /// further out than any millisecond an `i64` can hold.
    pub fn to_iso8601(self) -> String {
        OffsetDateTime::from_unix_timestamp_nanos(i128::from(self.0) * 1_000_000)
            .ok()
            .and_then(|t| t.format(&Rfc3339).ok())
            .unwrap_or_default()
    }
}

impl fmt::Display for Timestamp {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}ms", self.0)
    }
}

/// Reads a duration stored as integer milliseconds, such as `workflow_timeout_ms`.
///
/// `None` for a negative value, which no valid duration column holds.
pub fn duration_from_ms(ms: i64) -> Option<Duration> {
    u64::try_from(ms).ok().map(Duration::from_millis)
}

/// Reads a duration stored as fractional seconds, such as `polling_interval_sec`.
///
/// The other unit durations are stored in — see the module documentation.
///
/// `None` for anything that is not one: negative, NaN, infinite, or larger than [`Duration`] can
/// hold. The last is not hypothetical — the columns are `DOUBLE PRECISION`, so a row can carry
/// `1e300`, and `Duration::from_secs_f64` **panics** on it. Checking `is_finite()` does not cover
/// it; `try_from_secs_f64` covers all four.
pub fn duration_from_secs(secs: f64) -> Option<Duration> {
    Duration::try_from_secs_f64(secs).ok()
}

/// Rejects a workflow or fork's delay or timeout that is zero, or too long to store as
/// milliseconds, as [`Error::InvalidInput`] naming `field`.
///
/// The rule [`NewWorkflow::validate`] and [`ForkOptions::validate`] apply, kept in one place so
/// the surfaces that start a workflow can refuse the same values before they take a step id. A
/// zero delay is a caller that meant `None`, and a zero timeout would expire the workflow before
/// it ran. Both are written as milliseconds: the timeout as itself, the delay as an instant
/// resolved from it, so one that does not fit is refused rather than truncated into the column.
///
/// Neither rule reads a clock, so a surface can apply them before a step id is taken without a
/// replay deciding differently from the first run.
pub(crate) fn validate_positive_duration(
    field: &'static str,
    duration: Duration,
) -> Result<(), Error> {
    if duration.is_zero() {
        return Err(Error::InvalidInput {
            field: field.into(),
            detail: "must be a positive, non-zero duration".to_owned(),
        });
    }
    duration_to_ms(field, duration)?;
    Ok(())
}

/// Converts a caller's duration to the integer milliseconds a column such as
/// `workflow_timeout_ms` stores, or [`Error::InvalidInput`] naming `field` when it does not fit.
///
/// The inverse of [`duration_from_ms`], for the write side. `Duration::as_millis` is a `u128`, and
/// narrowing it with `as` would store a truncated, possibly negative, value in place of the one
/// the caller gave. Reported rather than saturated, for the reason stated on
/// [`Timestamp::to_system_time`]: a value this layer cannot store is the caller's to hear about,
/// not one to quietly replace with a different one.
pub fn duration_to_ms(field: &'static str, duration: Duration) -> Result<i64, Error> {
    i64::try_from(duration.as_millis()).map_err(|_| Error::InvalidInput {
        field: field.into(),
        detail: "must fit in milliseconds as a 64-bit integer".to_owned(),
    })
}

/// Where a workflow is in its lifecycle.
///
/// Stored as text and shared with the other SDKs on the same database, so the spellings are a
/// wire format rather than an internal choice.
///
/// The serde representation is those same spellings, which is what the rename on the derive is
/// for: a status that round-trips through a checkpoint — see [`DBOS::list_workflows`] — must
/// come back as what [`as_str`](Self::as_str) would have written.
///
/// [`DBOS::list_workflows`]: crate::DBOS::list_workflows
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
pub enum WorkflowStatus {
    /// Claimed by an executor and running.
    Pending,
    /// On a queue, waiting to be dequeued.
    Enqueued,
    /// On a queue but not yet eligible, because a delay or a debounce holds it back.
    Delayed,
    /// Finished, with an output.
    Success,
    /// Finished, with an error.
    Error,
    /// Cancelled before finishing.
    Cancelled,
    /// Recovered more times than allowed, and parked rather than retried again.
    MaxRecoveryAttemptsExceeded,
}

impl WorkflowStatus {
    /// The stored spelling.
    pub fn as_str(self) -> &'static str {
        match self {
            WorkflowStatus::Pending => "PENDING",
            WorkflowStatus::Enqueued => "ENQUEUED",
            WorkflowStatus::Delayed => "DELAYED",
            WorkflowStatus::Success => "SUCCESS",
            WorkflowStatus::Error => "ERROR",
            WorkflowStatus::Cancelled => "CANCELLED",
            WorkflowStatus::MaxRecoveryAttemptsExceeded => "MAX_RECOVERY_ATTEMPTS_EXCEEDED",
        }
    }

    /// Parses a stored value.
    ///
    /// Returns `None` for anything unrecognised rather than guessing. A status written by a
    /// newer SDK is a real possibility in a shared database, and silently mapping it
    /// onto the nearest known value would be worse than reporting that it is unknown.
    pub fn parse(value: &str) -> Option<Self> {
        Some(match value {
            "PENDING" => WorkflowStatus::Pending,
            "ENQUEUED" => WorkflowStatus::Enqueued,
            "DELAYED" => WorkflowStatus::Delayed,
            "SUCCESS" => WorkflowStatus::Success,
            "ERROR" => WorkflowStatus::Error,
            "CANCELLED" => WorkflowStatus::Cancelled,
            "MAX_RECOVERY_ATTEMPTS_EXCEEDED" => WorkflowStatus::MaxRecoveryAttemptsExceeded,
            _ => return None,
        })
    }

    /// Whether the workflow has finished and will not run again.
    pub fn is_terminal(self) -> bool {
        matches!(
            self,
            WorkflowStatus::Success | WorkflowStatus::Error | WorkflowStatus::Cancelled
        )
    }

    /// Whether the workflow is running or will run without anyone acting on it: `PENDING`,
    /// `ENQUEUED`, or `DELAYED` until its delay elapses.
    ///
    /// **Not the complement of [`is_terminal`](Self::is_terminal).**
    /// `MAX_RECOVERY_ATTEMPTS_EXCEEDED` is neither: it is parked rather than finished, and runs
    /// again only if someone resumes it.
    pub fn is_active(self) -> bool {
        matches!(
            self,
            WorkflowStatus::Pending | WorkflowStatus::Enqueued | WorkflowStatus::Delayed
        )
    }
}

impl fmt::Display for WorkflowStatus {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(self.as_str())
    }
}

/// What identifies a workflow, as the `workflow_status` row stores it.
///
/// Three columns rather than one name, because that is what Conductor and Console resolve against
/// and what a configured instance needs. Every SDK sharing the database stores these three
/// columns, so keying on them directly is the shape that reads back from any row.
///
/// **One spelling of absence.** The columns spell an absent class or config both as `NULL` and as
/// `''`, so every constructor stores an empty class or config as `None`. Two names for the same
/// workflow are therefore equal however they were built, and the fields are private so that holds.
///
/// A config with no class is a valid name, because a row can hold one, but nothing registers under
/// it: a configured instance belongs to a class. An empty name likewise names nothing.
/// [`DBOS::register_workflow`](crate::DBOS::register_workflow) refuses both.
#[derive(Clone, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
pub struct QualifiedWorkflowName {
    name: String,
    #[serde(default, deserialize_with = "absent_if_empty")]
    class_name: Option<String>,
    #[serde(default, deserialize_with = "absent_if_empty")]
    config_name: Option<String>,
}

impl QualifiedWorkflowName {
    /// A free function's name.
    pub fn new(name: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            class_name: None,
            config_name: None,
        }
    }

    /// A class-bound workflow's name, with no configured instance. An empty class is a free
    /// function's name.
    pub fn method(name: impl Into<String>, class_name: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            class_name: present(class_name.into()),
            config_name: None,
        }
    }

    /// A configured instance's name. An empty config is a method's name.
    pub fn instance(
        name: impl Into<String>,
        class_name: impl Into<String>,
        config_name: impl Into<String>,
    ) -> Self {
        Self {
            name: name.into(),
            class_name: present(class_name.into()),
            config_name: present(config_name.into()),
        }
    }

    /// Rebuilds a name from a status row's columns, treating `NULL` and `""` alike.
    ///
    /// Recovery is not why. A workflow belongs to one language from start to finish, so a row this
    /// executor recovers was written by this executor's own SDK and spells absence the one way
    /// Rust spells it. The boundary where a row arrives from *elsewhere* is **enqueue**: an
    /// application in another language can enqueue work for this one, and it writes the status row
    /// in its own spelling before any Rust executor sees it.
    ///
    /// Both spellings are in the wild: writers on the same database store an absent class or
    /// config as either, so a reader has to treat them alike.
    pub(crate) fn from_row(
        name: impl Into<String>,
        class_name: Option<&str>,
        config_name: Option<&str>,
    ) -> Self {
        Self {
            name: name.into(),
            class_name: class_name.map(str::to_owned).and_then(present),
            config_name: config_name.map(str::to_owned).and_then(present),
        }
    }

    /// The workflow's name.
    pub fn name(&self) -> &str {
        &self.name
    }

    /// The class a method-workflow belongs to; `None` for a free function.
    pub fn class_name(&self) -> Option<&str> {
        self.class_name.as_deref()
    }

    /// The configured instance this name is bound to; `None` unless instance-bound.
    pub fn config_name(&self) -> Option<&str> {
        self.config_name.as_deref()
    }
}

/// A stored class or config, with `""` read as absent, so a deserialized name holds the one
/// spelling of absence the constructors do.
fn absent_if_empty<'de, D>(deserializer: D) -> Result<Option<String>, D::Error>
where
    D: serde::Deserializer<'de>,
{
    let value: Option<String> = serde::Deserialize::deserialize(deserializer)?;
    Ok(value.and_then(present))
}

/// `Some` unless empty, the one spelling of absence a name holds.
fn present(value: String) -> Option<String> {
    (!value.is_empty()).then_some(value)
}

/// A free function's name, so a bare name stands for one wherever a [`QualifiedWorkflowName`] is
/// taken.
impl From<&str> for QualifiedWorkflowName {
    fn from(name: &str) -> Self {
        Self::new(name)
    }
}

/// A free function's name — see the `&str` conversion.
impl From<String> for QualifiedWorkflowName {
    fn from(name: String) -> Self {
        Self::new(name)
    }
}

/// A free function's name — see the `&str` conversion.
impl From<&String> for QualifiedWorkflowName {
    fn from(name: &String) -> Self {
        Self::new(name)
    }
}

/// A copy of a name held elsewhere, so a stored one can be passed by reference.
impl From<&QualifiedWorkflowName> for QualifiedWorkflowName {
    fn from(name: &QualifiedWorkflowName) -> Self {
        name.clone()
    }
}

impl WorkflowRecord {
    /// The name this row's workflow runs under, with its class and config, or `None` for a row
    /// that names no workflow.
    ///
    /// An empty class or config reads as absent, the same as `NULL`, which is how dequeue resolves
    /// a row against the registry. Pass it back to [`Client::enqueue`](crate::Client::enqueue) to
    /// run the same workflow again.
    pub fn qualified_name(&self) -> Option<QualifiedWorkflowName> {
        let name = self.name.as_deref().filter(|name| !name.is_empty())?;
        Some(QualifiedWorkflowName::from_row(
            name,
            self.class_name.as_deref(),
            self.config_name.as_deref(),
        ))
    }
}

/// `name`, `name/class` or `name/class/config`, for people to read. A config with no class keeps
/// its place as `name//config`, so it is never mistaken for a class.
impl std::fmt::Display for QualifiedWorkflowName {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match (&self.class_name, &self.config_name) {
            (Some(class), Some(config)) => write!(f, "{}/{class}/{config}", self.name),
            (Some(class), None) => write!(f, "{}/{class}", self.name),
            (None, Some(config)) => write!(f, "{}//{config}", self.name),
            (None, None) => f.write_str(&self.name),
        }
    }
}

/// A row of `workflow_status`.
///
/// **Shaped like the row, not like the domain.** Fields mirror columns one for one, including
/// their nullability: `name` is `Option` because the column is. Identity concepts that group
/// several columns — the `(name, class_name, config_name)` triple the registry keys on — are
/// built from a record where they are needed, rather than being baked into it. Grouping them
/// here would mean claiming `name` is non-null and quietly substituting an empty string when it
/// is not, which is the kind of loss this layer should report rather than absorb.
///
/// Every payload field holds encoded text — see the module documentation.
///
/// `Serialize`/`Deserialize` because a listing made from inside a workflow is checkpointed and
/// replays from what it recorded, so the rows have to round-trip. The derives give the record its
/// JSON shape, and the caller's [`StepCodec`] encodes that. No other SDK reads it — it
/// is the engine talking to its own replay.
///
/// **It is still a stored format, and the build that reads it is rarely the one that wrote it.**
/// A workflow checkpoints a listing on one deployment and replays it on the next, which is the
/// ordinary case rather than the exotic one — recovery after a deploy, and forking onto fixed
/// code, both land there. So **every field here carries `#[serde(default)]`** bar the two the
/// schema declares `NOT NULL` and this layer always writes, and a field added later must carry it
/// too. Without it, adding a column — this struct mirrors `workflow_status`, which gains them —
/// makes an older checkpoint unreadable, and `run_transactional_step` reports that as
/// [`Error::Malformed`]: not a retryable class, raised on the
/// replay path, so the workflow can never get past that step. Unknown fields are already
/// tolerated, `serde` ignoring them by default, so the reverse direction needs nothing.
///
/// For the same reason a field spelled differently from its column — `app_version`, `app_id`,
/// `app_name` — is serialized under the column's name, which is the key a recorded listing
/// already holds.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct WorkflowRecord {
    /// Primary key, and the identity a caller uses everywhere else.
    pub workflow_id: String,
    /// Lifecycle state.
    pub status: WorkflowStatus,
    /// Registered function name. Nullable in the schema, so nullable here.
    #[serde(default)]
    pub name: Option<String>,
    /// The type the function belongs to, for a method or a configured instance.
    #[serde(default)]
    pub class_name: Option<String>,
    /// The configured instance, if any.
    #[serde(default)]
    pub config_name: Option<String>,
    /// Encoded input, in whatever format `serialization` names.
    #[serde(default)]
    pub input: Option<String>,
    /// Encoded output, present once the workflow succeeds.
    #[serde(default)]
    pub output: Option<String>,
    /// Encoded error, present once the workflow fails.
    #[serde(default)]
    pub error: Option<String>,
    /// Which format the payloads above are in. `None` on rows written before the column
    /// existed, which means the writer's own default.
    #[serde(default)]
    pub serialization: Option<String>,
    /// The executor that most recently claimed this workflow.
    #[serde(default)]
    pub executor_id: Option<String>,
    /// Application version that created it, used to keep recovery on compatible code.
    #[serde(default, rename = "application_version")]
    pub app_version: Option<String>,
    /// Recovery attempts so far, against the dead-letter limit.
    #[serde(default)]
    pub recovery_attempts: i64,
    /// Queue this workflow was enqueued on, if it was.
    #[serde(default)]
    pub queue_name: Option<String>,
    /// When the row was created.
    #[serde(default)]
    pub created_at: Timestamp,
    /// When the row last changed.
    #[serde(default)]
    pub updated_at: Timestamp,
    /// When execution began, if it has.
    #[serde(default)]
    pub started_at: Option<Timestamp>,
    /// When the workflow reached a terminal state.
    #[serde(default)]
    pub completed_at: Option<Timestamp>,
    /// The workflow that forked this one, if any.
    #[serde(default)]
    pub forked_from: Option<String>,
    /// The workflow that started this one as a child, if any.
    #[serde(default)]
    pub parent_workflow_id: Option<String>,
    /// Whether this workflow has been forked from at least once.
    #[serde(default)]
    pub was_forked_from: bool,

    // ── Ownership and attribution ──────────────────────────────────────────────
    /// Identity of the attempt currently holding this workflow.
    ///
    /// The single-execution guard: distinct per attempt, unlike `executor_id`, which defaults
    /// to `"local"` and collides between processes on one machine.
    #[serde(default)]
    pub owner_xid: Option<String>,
    /// Deployment identifier, for installations running several applications.
    #[serde(default, rename = "application_id")]
    pub app_id: Option<String>,
    /// User on whose behalf the workflow runs.
    #[serde(default)]
    pub authenticated_user: Option<String>,
    /// Roles that user holds.
    ///
    /// Decoded from the column's JSON array, which this layer owns — see
    /// [`NewWorkflow::authenticated_roles`]. A NULL column reads as empty.
    #[serde(default)]
    pub authenticated_roles: Vec<String>,
    /// Role actually assumed for this execution.
    #[serde(default)]
    pub assumed_role: Option<String>,
    /// Request context captured at creation.
    #[serde(default)]
    pub request: Option<String>,
    /// The application that owns this workflow, or `None` if it is unclaimed.
    ///
    /// Unclaimed means no application has taken it — a row written before ownership existed, or by
    /// a handle with no application of its own. Every application may run it, and the first to
    /// dequeue it claims it.
    #[serde(default, rename = "application_name")]
    pub app_name: Option<String>,

    // ── Queueing ───────────────────────────────────────────────────────────────
    /// Deduplication key within the queue. At most one live workflow may hold a given key.
    #[serde(default)]
    pub deduplication_id: Option<String>,
    /// Dequeue priority; lower runs sooner.
    ///
    /// `i32` and not optional, mirroring `INT4 NOT NULL DEFAULT 0`. Widening to `i64` would let
    /// values round-trip through a type the column cannot hold, and making it optional would
    /// invite writing a NULL the column rejects.
    #[serde(default)]
    pub priority: i32,
    /// Partition this workflow belongs to, on a partitioned queue.
    #[serde(default)]
    pub queue_partition_key: Option<String>,
    /// Whether a rate limiter is currently holding this workflow back.
    #[serde(default)]
    pub rate_limited: bool,
    /// Schedule that enqueued this workflow, if a schedule did.
    #[serde(default)]
    pub schedule_name: Option<String>,

    // ── Timing ─────────────────────────────────────────────────────────────────
    /// How long the workflow may run for.
    ///
    /// A **duration**, unlike `deadline` below, which is an instant. The two are adjacent
    /// columns and mean different things: this is a budget, that is a wall-clock cutoff.
    #[serde(default)]
    pub timeout: Option<Duration>,
    /// Wall-clock instant the workflow must finish by.
    #[serde(default)]
    pub deadline: Option<Timestamp>,
    /// Instant before which the workflow must not be dequeued.
    #[serde(default)]
    pub delay_until: Option<Timestamp>,
    /// Cap past which a debounce may not push the delay any further.
    #[serde(default)]
    pub debounce_deadline: Option<Timestamp>,
    /// Whether the deduplication id is a debounce key, cleared on DELAYED to ENQUEUED.
    #[serde(default)]
    pub is_debounced: bool,

    // ── Caller-supplied metadata ───────────────────────────────────────────────
    /// Arbitrary attributes attached at creation, stored as JSON.
    ///
    /// Opaque here like every other payload: this layer stores and returns the text and does
    /// not parse it, even though the column is `jsonb` and is queried by containment.
    #[serde(default)]
    pub attributes: Option<String>,
}

/// What a start or an enqueue does when its workflow id already names a workflow.
///
/// The id is an idempotency key, so the default joins: the caller gets a handle to the workflow
/// already there, running or finished, and its own input is discarded. [`Reject`](Self::Reject)
/// is for an id that must be new, such as one derived from a payment that may only be processed
/// once.
///
/// Only another caller's workflow counts. A start that retries after its first attempt's commit
/// was lost finds its own row, and is neither joined nor refused.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum WorkflowIdReusePolicy {
    /// Hand back a handle to the workflow that already has the id.
    #[default]
    ReturnExisting,
    /// Refuse the start with [`Error::WorkflowIdInUse`], whatever the existing workflow's status,
    /// writing nothing. Only for an id the caller chose: a start or an enqueue that names none is
    /// refused.
    ///
    /// A parent's own child, started at a step a rewind has since erased from the parent's
    /// record, is not a reuse: the replayed parent adopts it.
    ///
    /// **Started from inside a workflow, the refusal is recorded at the parent's step**, so a
    /// replay of the parent gets the same error and starts nothing.
    Reject,
}

/// A workflow being created, as opposed to one being read back.
///
/// This is deliberately *not* [`WorkflowRecord`]. A row has 37 columns; a caller creating a
/// workflow can meaningfully set 24 of them. The rest are the database's to write:
///
/// - `status` and `recovery_attempts` are **derived** from the queue and delay below, so a
///   caller cannot enqueue a workflow and then label it `SUCCESS`.
/// - `created_at`, `updated_at`, and `owner_xid` are stamped at insert.
/// - `output`, `error`, `started_at`, `completed_at`, `forked_from`, `was_forked_from`, and
///   `rate_limited` belong to execution, forking, and the rate limiter. A workflow that has
///   not started has no output to offer.
/// - `parent_workflow_id` belongs to the child start, and travels on [`InitWorkflowCaller`] with
///   the rest of what the parent contributes. A workflow has a parent exactly when its start was
///   recorded against one — the two are one event, and one of them writing without the other is
///   the state the pair exists to make unreachable, so there is no caller who can meaningfully
///   set the column alone.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct NewWorkflow<'a> {
    /// The id, which is what makes a retried submission the same workflow.
    pub workflow_id: &'a str,
    /// The registered function name.
    pub name: Option<&'a str>,
    /// The class the function belongs to, for class-bound workflows.
    pub class_name: Option<&'a str>,
    /// The configured instance name, for instance-bound workflows.
    pub config_name: Option<&'a str>,
    /// Encoded arguments.
    pub input: Option<&'a str>,
    /// How `input` and, later, the outcome are encoded.
    pub serialization: Option<&'a str>,

    /// The queue to enqueue on. `None` runs the workflow directly.
    ///
    /// This decides the initial status: no queue means `PENDING`, a queue means `ENQUEUED`, and
    /// a queue with a `delay` means `DELAYED`.
    pub queue_name: Option<&'a str>,
    /// Deduplication key within the queue.
    pub deduplication_id: Option<&'a str>,
    /// Dequeue priority; lower runs sooner.
    pub priority: i32,
    /// Partition within the queue.
    pub queue_partition_key: Option<&'a str>,
    /// How long to hold the workflow before it becomes eligible to dequeue.
    ///
    /// A duration, not an instant: the insert resolves it against the database clock it writes
    /// `created_at` with, the clock the release is decided by. Letting the caller compute
    /// `now + delay` would put its clock skew into the row.
    pub delay: Option<Duration>,
    /// Marks `deduplication_id` as a debounce key to clear on the `DELAYED` → `ENQUEUED` move.
    ///
    /// A debounced row must be new: an existing workflow under the id is
    /// [`Error::ConflictingWorkflow`] rather than joined.
    pub is_debounced: bool,
    /// How far past the insert a debounce may push the release: the debounce deadline, as a
    /// duration.
    ///
    /// Resolved by the insert against the same clock reading as `delay`, which it also caps, so a
    /// delay longer than the timeout is released at the deadline exactly.
    pub debounce_timeout: Option<Duration>,

    /// Wall-clock budget for the whole workflow.
    ///
    /// A workflow inserted `PENDING` with no `deadline` gets one from this, on the database's
    /// clock as the insert runs; a queued or delayed one gets it when a dequeue claims it.
    pub timeout: Option<Duration>,
    /// Absolute expiry, when the parent already fixed one.
    ///
    /// Distinct from `timeout` because a child inherits its parent's deadline rather than
    /// restarting the clock. Kept as given, whatever `timeout` says.
    pub deadline: Option<Timestamp>,

    /// The executor claiming this workflow.
    pub executor_id: Option<&'a str>,

    /// The application this workflow belongs to; `None` means the writing handle's own.
    ///
    /// On the input rather than taken from the handle for the same reason
    /// [`executor_id`](Self::executor_id) is: a client enqueueing *for* another application names
    /// it here, and only the insert decides a workflow's owner — nothing re-owns it afterwards.
    /// Leaving both this and the handle's name unset writes an unclaimed workflow, which every
    /// application may run and the first to dequeue claims.
    pub application_name: Option<&'a str>,
    /// Application version, which recovery uses to avoid resuming under changed code.
    pub application_version: Option<&'a str>,
    /// Application id, as assigned by the platform.
    pub application_id: Option<&'a str>,

    /// Authenticated principal at submission.
    pub authenticated_user: Option<&'a str>,
    /// The roles that principal held.
    ///
    /// A list rather than an encoded string, unlike `input` and the outcome payloads: those are
    /// opaque to this layer and cross as whatever the caller encoded, but the column is *always*
    /// a JSON array of strings, a format shared with the other SDKs that read the column, so the
    /// encoding belongs here.
    ///
    /// Empty is stored as NULL, on the same reasoning as the empty-string normalisation for
    /// `authenticated_user`: "no roles" should have one representation in the column.
    pub authenticated_roles: Vec<&'a str>,
    /// The role actually assumed.
    pub assumed_role: Option<&'a str>,

    /// The schedule that triggered this workflow. Set only by the scheduler.
    pub schedule_name: Option<&'a str>,
    /// Caller-supplied JSON attributes, stored in a `jsonb` column.
    pub attributes: Option<&'a str>,
}

impl<'a> NewWorkflow<'a> {
    /// A workflow with an id and nothing else set.
    pub fn new(workflow_id: &'a str) -> Self {
        Self {
            workflow_id,
            ..Self::default()
        }
    }

    /// Rejects values the system database will not store.
    ///
    /// `NewWorkflow` has public fields and so has no construction hook to refuse a bad value; this
    /// runs at the point of use instead.
    ///
    /// **An empty string is not a missing value — `None` is.** A `Some("")` queue name would
    /// enqueue a workflow onto a queue called `""`, and an empty `workflow_id` would create a
    /// row no caller can ever name again. Both are caller bugs worth reporting rather than
    /// storing.
    ///
    /// The two auth fields are exempt: `""` is a common way for a caller to say there is no auth
    /// context, so those are normalised at write time instead of rejected.
    pub fn validate(&self) -> Result<(), Error> {
        for (field, value) in [("workflow_id", Some(self.workflow_id)), ("name", self.name)] {
            if value == Some("") {
                return Err(Error::InvalidInput {
                    field: field.into(),
                    detail: "must not be empty".to_owned(),
                });
            }
        }
        for (field, value) in [
            ("class_name", &self.class_name),
            ("config_name", &self.config_name),
            ("queue_name", &self.queue_name),
            ("deduplication_id", &self.deduplication_id),
            ("queue_partition_key", &self.queue_partition_key),
            ("schedule_name", &self.schedule_name),
            ("input", &self.input),
            ("serialization", &self.serialization),
            ("application_version", &self.application_version),
        ] {
            if *value == Some("") {
                return Err(Error::InvalidInput {
                    field: field.into(),
                    detail: "must be absent rather than empty".to_owned(),
                });
            }
        }
        // `Duration` already rules out negatives, so zero and a value too long to store are what
        // is left; `validate_positive_duration` says why each is refused.
        for (field, value) in [
            ("delay", self.delay),
            ("timeout", self.timeout),
            ("debounce_timeout", self.debounce_timeout),
        ] {
            if let Some(value) = value {
                validate_positive_duration(field, value)?;
            }
        }
        validate_attributes(self.attributes)?;
        Ok(())
    }

    /// The status this workflow starts in, which follows from the queue and the delay.
    ///
    /// Not a caller's choice: a workflow is `PENDING` when it runs here, `ENQUEUED` when it waits
    /// on a queue, and `DELAYED` when it waits on a queue and a clock.
    pub fn initial_status(&self) -> WorkflowStatus {
        match (&self.queue_name, &self.delay) {
            (None, _) => WorkflowStatus::Pending,
            (Some(_), None) => WorkflowStatus::Enqueued,
            (Some(_), Some(_)) => WorkflowStatus::Delayed,
        }
    }
}

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

    /// Every recorded step name, spelled out. A replay compares these exactly, so renaming one
    /// strands every workflow recorded under the old name; this test is where such a rename has to
    /// be made on purpose. The names Rust 0.5 recorded are among them, unchanged.
    #[test]
    fn the_step_names_are_pinned() {
        use step_names::*;
        let pinned = [
            (SLEEP, "DBOS.sleep"),
            (PATCH_PREFIX, "DBOS.patch-"),
            (SEND, "DBOS.send"),
            (SEND_BULK, "DBOS.sendBulk"),
            (WRITE_STREAM, "DBOS.writeStream"),
            (CLOSE_STREAM, "DBOS.closeStream"),
            (READ_STREAM, "DBOS.readStream"),
            (READ_STREAM_VALUE, "DBOS.readStreamValue"),
            (SET_EVENT, "DBOS.setEvent"),
            (GET_EVENT, "DBOS.getEvent"),
            (GET_RESULT, "DBOS.getResult"),
            (SELECT_WORKFLOW, "DBOS.selectWorkflow"),
            (SELECT_STEP, "DBOS.selectStep"),
            (RECV, "DBOS.recv"),
            (DEBOUNCE, "DBOS.debounceDelayedWorkflow"),
            (CANCEL_WORKFLOW, "DBOS.cancelWorkflow"),
            (RESUME_WORKFLOW, "DBOS.resumeWorkflow"),
            (DELETE_WORKFLOW, "DBOS.deleteWorkflow"),
            (FORK_WORKFLOW, "DBOS.forkWorkflow"),
            (REWIND_WORKFLOW, "DBOS.rewindWorkflow"),
            (SET_WORKFLOW_DELAY, "DBOS.setWorkflowDelay"),
            (UPDATE_WORKFLOW_ATTRIBUTES, "DBOS.updateWorkflowAttributes"),
            (LIST_WORKFLOWS, "DBOS.listWorkflows"),
            (LIST_WORKFLOW_STEPS, "DBOS.listWorkflowSteps"),
            (CREATE_SCHEDULE, "DBOS.createSchedule"),
            (UPSERT_SCHEDULE, "DBOS.upsertSchedule"),
            (GET_SCHEDULE, "DBOS.getSchedule"),
            (LIST_SCHEDULES, "DBOS.listSchedules"),
            (UPDATE_SCHEDULE, "DBOS.updateSchedule"),
            (PAUSE_SCHEDULE, "DBOS.pauseSchedule"),
            (RESUME_SCHEDULE, "DBOS.resumeSchedule"),
            (DELETE_SCHEDULE, "DBOS.deleteSchedule"),
        ];
        for (constant, literal) in pinned {
            assert_eq!(constant, literal);
        }
    }

    /// Every way of building a name stores an empty class or config as absent, so names for the
    /// same workflow are equal however they were built.
    #[test]
    fn a_name_holds_one_spelling_of_absence() {
        let free = QualifiedWorkflowName::new("f");
        assert_eq!(QualifiedWorkflowName::method("f", ""), free);
        assert_eq!(QualifiedWorkflowName::instance("f", "", ""), free);
        assert_eq!(
            QualifiedWorkflowName::from_row("f", Some(""), Some("")),
            free
        );
        assert_eq!(
            QualifiedWorkflowName::instance("f", "C", ""),
            QualifiedWorkflowName::method("f", "C")
        );

        let unclassed = QualifiedWorkflowName::instance("f", "", "eu");
        assert_eq!(
            unclassed,
            QualifiedWorkflowName::from_row("f", None, Some("eu"))
        );
        assert_eq!(unclassed.class_name(), None);
        assert_eq!(unclassed.config_name(), Some("eu"));
    }

    /// The display keeps a config with no class apart from a class.
    #[test]
    fn a_name_displays_each_part_in_its_place() {
        assert_eq!(QualifiedWorkflowName::new("f").to_string(), "f");
        assert_eq!(QualifiedWorkflowName::method("f", "C").to_string(), "f/C");
        assert_eq!(
            QualifiedWorkflowName::instance("f", "C", "eu").to_string(),
            "f/C/eu"
        );
        assert_eq!(
            QualifiedWorkflowName::instance("f", "", "eu").to_string(),
            "f//eu"
        );
    }

    /// A stored name that spells absence `""` reads back as one that spells it `None`.
    #[test]
    fn a_deserialized_name_holds_one_spelling_of_absence() {
        let key: QualifiedWorkflowName =
            serde_json::from_str(r#"{"name":"f","class_name":"","config_name":null}"#).unwrap();
        assert_eq!(key, QualifiedWorkflowName::new("f"));
        let bare: QualifiedWorkflowName = serde_json::from_str(r#"{"name":"f"}"#).unwrap();
        assert_eq!(
            bare,
            QualifiedWorkflowName::new("f"),
            "a missing part is absent"
        );
        let back: QualifiedWorkflowName = serde_json::from_str(
            &serde_json::to_string(&QualifiedWorkflowName::instance("f", "C", "eu")).unwrap(),
        )
        .unwrap();
        assert_eq!(back, QualifiedWorkflowName::instance("f", "C", "eu"));
    }

    /// The millisecond conversion either side of `time`, which is the part this crate owns.
    #[test]
    fn an_instant_formats_as_iso8601() {
        let iso = |ms| Timestamp::from_epoch_ms(ms).to_iso8601();

        assert_eq!(iso(0), "1970-01-01T00:00:00Z");
        assert_eq!(iso(1_786_492_800_000), "2026-08-12T00:00:00Z");
        // A fraction appears only when there is one.
        assert_eq!(iso(1_786_492_800_123), "2026-08-12T00:00:00.123Z");
        // Before the epoch the millisecond floors rather than truncating towards zero, which is
        // the direction `i128::div_euclid` is chosen for.
        assert_eq!(iso(-1), "1969-12-31T23:59:59.999Z");
        assert_eq!(iso(-86_400_000), "1969-12-31T00:00:00Z");
    }

    /// The spellings writers on a shared database use all read back as the same instant.
    #[test]
    fn an_iso8601_instant_parses_from_every_writers_format() {
        let parse = Timestamp::parse_iso8601;
        let expected = Some(Timestamp::from_epoch_ms(1_786_492_800_000));

        assert_eq!(
            parse("2026-08-12T00:00:00.000Z"),
            expected,
            "milliseconds, Z"
        );
        assert_eq!(parse("2026-08-12T00:00:00Z"), expected, "bare Z");
        assert_eq!(
            parse("2026-08-12T00:00:00+00:00"),
            expected,
            "+00:00 offset"
        );
        assert_eq!(
            parse("2026-08-12T00:00:00.000000000Z"),
            expected,
            "RFC3339Nano"
        );
        // An offset is applied rather than ignored, in either direction.
        assert_eq!(parse("2026-08-12T01:30:00+01:30"), expected);
        assert_eq!(parse("2026-08-11T19:00:00-05:00"), expected);

        // Conductor formats with an explicit numeric offset rather than `Z`, and elides the
        // fraction on a whole second, so both of these reach a reader.
        assert_eq!(parse("2026-08-12T00:00:00+00:00"), expected);
        assert_eq!(
            parse("2026-08-12T00:00:00.123+00:00"),
            Some(Timestamp::from_epoch_ms(1_786_492_800_123))
        );

        // Sub-millisecond digits truncate rather than round, and a short fraction pads.
        let ms = |t: Option<Timestamp>| t.unwrap().as_epoch_ms() % 1_000;
        assert_eq!(ms(parse("2026-08-12T00:00:00.5Z")), 500);
        assert_eq!(ms(parse("2026-08-12T00:00:00.123999Z")), 123);

        // Strict RFC 3339: the offset is required and carries its colon. Neither rejection is
        // reachable from a writer, which always emits an offset with its colon, and Conductor's
        // own reader rejects exactly the same two. Accepting them would be this crate inventing
        // leniency nobody asked for.
        for bad in [
            "",
            "2026-08-12",
            "2026-08-12T00:00",
            "not-a-date-at-all",
            "2026-13-01T00:00:00Z",
            "2026-08-12T24:00:00Z",
            "2026-08-12T00:00:00",
            "2026-08-12T00:00:00+0130",
        ] {
            assert_eq!(parse(bad), None, "{bad:?} is not an instant this reads");
        }
    }

    /// Formatting and parsing are inverses, which is what makes the column round-trip.
    #[test]
    fn an_iso8601_instant_round_trips() {
        for ms in [
            0,
            -1,
            1_786_492_800_000,
            1_709_209_845_123,
            951_827_445_999,
            -2_203_977_600_000,
        ] {
            let instant = Timestamp::from_epoch_ms(ms);
            assert_eq!(
                Timestamp::parse_iso8601(&instant.to_iso8601()),
                Some(instant),
                "{ms} did not survive the round trip"
            );
        }
    }

    /// Instants and durations do not mix, which is the point of separating them.
    #[test]
    fn a_deadline_is_a_start_plus_a_timeout() {
        let start = Timestamp::from_epoch_ms(1_700_000_000_000);
        let timeout = Duration::from_secs(30);
        let deadline = start.checked_add(timeout).unwrap();

        assert_eq!(deadline.as_epoch_ms(), 1_700_000_030_000);
        assert_eq!(deadline.duration_since(start), Some(timeout));
        // Going backwards is not a duration.
        assert_eq!(start.duration_since(deadline), None);
    }

    #[test]
    fn timestamps_round_trip_through_system_time() {
        let t = Timestamp::from_epoch_ms(1_700_000_000_123);
        let round_tripped = Timestamp::from_system_time(t.to_system_time().unwrap()).unwrap();
        assert_eq!(round_tripped, t);
    }

    /// An instant this layer cannot store is reported, whichever side of the epoch it falls.
    #[test]
    fn an_unstorable_system_time_is_not_a_timestamp() {
        assert_eq!(
            Timestamp::from_system_time(UNIX_EPOCH - Duration::from_secs(1)),
            None,
            "before the epoch"
        );
        // Representable as a `SystemTime`, but not as milliseconds in an `i64`.
        let far = UNIX_EPOCH
            .checked_add(Duration::from_secs(i64::MAX as u64 / 100))
            .expect("a SystemTime that far out is constructible");
        assert_eq!(Timestamp::from_system_time(far), None, "too far after it");
    }

    /// A negative instant is reported rather than silently becoming 1970.
    #[test]
    fn a_pre_epoch_instant_does_not_convert() {
        assert_eq!(Timestamp::from_epoch_ms(-1).to_system_time(), None);
    }

    /// The two units durations are stored in, read through their own helpers.
    #[test]
    fn durations_read_from_either_stored_unit() {
        assert_eq!(duration_from_ms(1_500), Some(Duration::from_millis(1_500)));
        assert_eq!(duration_from_secs(1.5), Some(Duration::from_millis(1_500)));
        // Neither unit has a meaningful negative, and a float column can be NaN.
        assert_eq!(duration_from_ms(-1), None);
        assert_eq!(duration_from_secs(-1.0), None);
        assert_eq!(duration_from_secs(f64::NAN), None);
        assert_eq!(duration_from_secs(f64::INFINITY), None);
        // Finite and non-negative, and still not a duration: an `is_finite()` guard alone lets it
        // through to a panic, and the columns it reads are wide enough to hold it.
        assert_eq!(duration_from_secs(1e300), None);
    }

    /// A duration converts while its milliseconds fit an `i64`, and is refused one past that.
    #[test]
    fn a_duration_converts_to_milliseconds_only_while_it_fits() {
        let largest = Duration::from_millis(i64::MAX as u64);
        assert_eq!(duration_to_ms("timeout", largest).unwrap(), i64::MAX);
        assert_eq!(duration_to_ms("timeout", Duration::ZERO).unwrap(), 0);
        // Sub-millisecond precision is dropped, as the column cannot hold it, and is not what
        // pushes a value over.
        let largest_with_spare_nanos = largest + Duration::from_nanos(999_999);
        assert_eq!(
            duration_to_ms("timeout", largest_with_spare_nanos).unwrap(),
            i64::MAX
        );

        for refused in [largest + Duration::from_millis(1), Duration::MAX] {
            match duration_to_ms("timeout", refused) {
                Err(Error::InvalidInput { field, detail }) => {
                    assert_eq!(field, "timeout");
                    assert_eq!(detail, "must fit in milliseconds as a 64-bit integer");
                }
                other => panic!("{refused:?} should have been refused, got {other:?}"),
            }
        }
    }

    /// An instant plus a caller's duration is refused, not wrapped, past the largest instant.
    #[test]
    fn an_instant_past_the_largest_is_refused_not_wrapped() {
        let start = Timestamp::from_epoch_ms(1_700_000_000_000);
        assert_eq!(
            start.try_add("delay", Duration::from_secs(1)).unwrap(),
            Timestamp::from_epoch_ms(1_700_000_001_000)
        );
        let room = Duration::from_millis((i64::MAX - start.as_epoch_ms()) as u64);
        assert_eq!(
            start.try_add("delay", room).unwrap(),
            Timestamp::from_epoch_ms(i64::MAX)
        );

        // Fits in milliseconds on its own, and still overflows the sum.
        let over = room + Duration::from_millis(1);
        assert!(matches!(
            start.try_add("delay", over),
            Err(Error::InvalidInput { ref field, .. }) if field == "delay"
        ));
        // Does not fit at all.
        assert!(matches!(
            start.try_add("duration", Duration::MAX),
            Err(Error::InvalidInput { ref field, .. }) if field == "duration"
        ));
    }

    /// A bounce's period is held to the rule a delay is: positive, and storable as milliseconds.
    #[test]
    fn a_bounce_refuses_a_zero_or_unstorable_period() {
        let workflow = QualifiedWorkflowName::new("wf");
        let request = |period| DebounceRequest {
            workflow: &workflow,
            queue_name: "q",
            deduplication_id: "key",
            period,
            inputs: None,
            serialization: None,
            application_name: None,
        };
        assert!(request(Duration::from_secs(1)).validate().is_ok());
        for refused in [Duration::ZERO, Duration::MAX] {
            assert!(
                matches!(
                    request(refused).validate(),
                    Err(Error::InvalidInput { ref field, .. }) if field == "period"
                ),
                "{refused:?} should be refused"
            );
        }
    }

    /// A workflow's timeout, delay and debounce timeout are refused by validation when they cannot
    /// be stored, so the refusal comes before anything is written.
    #[test]
    fn an_oversized_workflow_duration_does_not_validate() {
        let fits = NewWorkflow {
            debounce_timeout: Some(Duration::from_millis(i64::MAX as u64)),
            delay: Some(Duration::from_millis(i64::MAX as u64)),
            timeout: Some(Duration::from_millis(i64::MAX as u64)),
            ..NewWorkflow::new("wf")
        };
        assert!(fits.validate().is_ok());

        let timeout = NewWorkflow {
            timeout: Some(Duration::MAX),
            ..NewWorkflow::new("wf")
        };
        assert!(matches!(
            timeout.validate(),
            Err(Error::InvalidInput { ref field, .. }) if field == "timeout"
        ));
        let delay = NewWorkflow {
            queue_name: Some("q"),
            delay: Some(Duration::MAX),
            ..NewWorkflow::new("wf")
        };
        assert!(matches!(
            delay.validate(),
            Err(Error::InvalidInput { ref field, .. }) if field == "delay"
        ));
        let debounce_timeout = NewWorkflow {
            queue_name: Some("q"),
            delay: Some(Duration::from_secs(1)),
            debounce_timeout: Some(Duration::MAX),
            ..NewWorkflow::new("wf")
        };
        assert!(matches!(
            debounce_timeout.validate(),
            Err(Error::InvalidInput { ref field, .. }) if field == "debounce_timeout"
        ));
    }

    /// Both readers accept zero, and for the same reason.
    ///
    /// Zero is a legitimate stored value — these read what a column holds, and rejecting it
    /// here would turn a readable row into an error. The rule that a duration *option* must be
    /// strictly positive belongs at the input boundary, where a caller supplies one, not at the
    /// point a stored value is read back.
    ///
    /// `-0.0` counts as zero: it compares equal to `0.0`, which is what the guard tests.
    #[test]
    fn both_readers_accept_zero() {
        assert_eq!(duration_from_ms(0), Some(Duration::ZERO));
        assert_eq!(duration_from_secs(0.0), Some(Duration::ZERO));
        assert_eq!(duration_from_secs(-0.0), Some(Duration::ZERO));
    }

    /// The spellings are a wire format shared with the other SDKs on the same database.
    #[test]
    fn status_round_trips_through_its_stored_spelling() {
        for status in [
            WorkflowStatus::Pending,
            WorkflowStatus::Enqueued,
            WorkflowStatus::Delayed,
            WorkflowStatus::Success,
            WorkflowStatus::Error,
            WorkflowStatus::Cancelled,
            WorkflowStatus::MaxRecoveryAttemptsExceeded,
        ] {
            assert_eq!(WorkflowStatus::parse(status.as_str()), Some(status));
        }
    }

    /// Pinned, because other SDKs sharing the database store and read these exact strings.
    #[test]
    fn stored_spellings_are_the_shared_wire_strings() {
        assert_eq!(WorkflowStatus::Pending.as_str(), "PENDING");
        assert_eq!(WorkflowStatus::Enqueued.as_str(), "ENQUEUED");
        assert_eq!(WorkflowStatus::Delayed.as_str(), "DELAYED");
        assert_eq!(WorkflowStatus::Success.as_str(), "SUCCESS");
        assert_eq!(WorkflowStatus::Error.as_str(), "ERROR");
        assert_eq!(WorkflowStatus::Cancelled.as_str(), "CANCELLED");
        assert_eq!(
            WorkflowStatus::MaxRecoveryAttemptsExceeded.as_str(),
            "MAX_RECOVERY_ATTEMPTS_EXCEEDED",
        );
    }

    /// An unknown status is reported as unknown, not mapped onto a neighbour.
    #[test]
    fn an_unrecognised_status_does_not_parse() {
        assert_eq!(WorkflowStatus::parse("SOMETHING_NEWER"), None);
        assert_eq!(WorkflowStatus::parse("pending"), None, "matching is exact");
    }

    #[test]
    fn terminal_states_are_the_three_that_stop_execution() {
        assert!(WorkflowStatus::Success.is_terminal());
        assert!(WorkflowStatus::Error.is_terminal());
        assert!(WorkflowStatus::Cancelled.is_terminal());
        assert!(!WorkflowStatus::Pending.is_terminal());
        assert!(!WorkflowStatus::Enqueued.is_terminal());
        assert!(!WorkflowStatus::Delayed.is_terminal());
        // Parked rather than finished: it can still be resumed.
        assert!(!WorkflowStatus::MaxRecoveryAttemptsExceeded.is_terminal());
    }
}

/// Which applications' rows a query covers.
///
/// Three cases rather than a list, because "no applications named" is genuinely ambiguous and the
/// two readings are opposites: an *empty* list of applications could mean every application, and an
/// *absent* one this handle's own. Spelled out here so it cannot be collapsed by a later tidy-up
/// that sees an empty list and a missing one as the same thing.
///
/// Unclaimed rows are matched in every case: they belong to no application, so they belong to all
/// of them. See [`AppScope::Named`] for the one exception that is not.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub enum AppScope<'a> {
    /// Let the query decide, which is what a caller who has not thought about it wants.
    ///
    /// A **search** scopes to the handle's own application: listing workflows without saying whose
    /// should not return a peer's. An **id-keyed read** matches every application instead, because
    /// a workflow id is a global address — asking for one by id is an identity read, not a search,
    /// and answering "no such workflow" for one that plainly exists would be a lie.
    #[default]
    Unset,
    /// Every application, said deliberately. What an operator's cross-application view asks for.
    Any,
    /// These applications, plus the unclaimed rows.
    ///
    /// An empty list is [`AppScope::Any`] rather than "no applications": there is no useful
    /// query for rows belonging to none of the applications you named, and reading it as a
    /// narrowing would make an unfiltered UI silently show nothing.
    Named(Vec<&'a str>),
}

/// Which workflows to list, and how much of each to load.
///
/// Every field is a narrowing, and the default narrows nothing — so
/// `WorkflowFilter::default()` lists everything. List fields match any of their entries and are
/// ignored when empty; `Option<bool>` fields are three-valued, where `None` does not filter.
///
/// [`app_scope`](Self::app_scope) is the exception, and deliberately so: its default scopes
/// to the caller's own application rather than to everything.
///
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WorkflowFilter<'a> {
    /// Exact workflow ids.
    ///
    /// Supplying any makes this an id-keyed read, which changes what
    /// [`app_scope`](Self::app_scope) defaults to — see [`AppScope::Unset`].
    pub workflow_ids: Vec<&'a str>,
    /// Workflow ids starting with any of these.
    ///
    /// A prefix is a search, not an address, so unlike [`workflow_ids`](Self::workflow_ids) it
    /// does not make this an id-keyed read.
    pub workflow_id_prefixes: Vec<&'a str>,

    /// Whose workflows to list. Defaults to the caller's own application plus unclaimed ones.
    pub app_scope: AppScope<'a>,

    /// Registered function names.
    pub names: Vec<&'a str>,
    /// Class names, for class-bound workflows.
    pub class_names: Vec<&'a str>,
    /// Configured instance names.
    pub config_names: Vec<&'a str>,
    /// Statuses to include.
    pub status: Vec<WorkflowStatus>,

    /// Application versions.
    pub app_versions: Vec<&'a str>,
    /// Executors that claimed the workflow.
    pub executor_ids: Vec<&'a str>,
    /// Authenticated principals at submission.
    pub authenticated_users: Vec<&'a str>,

    /// Queues the workflow was submitted to.
    pub queue_names: Vec<&'a str>,
    /// Only workflows that went through a queue at all.
    pub queues_only: bool,
    /// Schedules that triggered the workflow.
    pub schedule_names: Vec<&'a str>,
    /// Deduplication keys.
    pub deduplication_ids: Vec<&'a str>,
    /// Whether the deduplication key is a debounce key.
    pub is_debounced: Option<bool>,

    /// Workflows started by any of these.
    pub parent_workflow_ids: Vec<&'a str>,
    /// Whether the workflow has a parent at all.
    pub has_parent: Option<bool>,
    /// Workflows forked from any of these.
    pub forked_from: Vec<&'a str>,
    /// Whether the workflow is itself a fork, i.e. whether it has a `forked_from`.
    pub is_fork: Option<bool>,
    /// Whether this workflow was itself forked from another.
    pub was_forked_from: Option<bool>,

    /// Created at or after this instant.
    ///
    /// Not `start_time`, which would read as "when the workflow started": it filters
    /// `created_at`, and a workflow may be created long before it starts.
    /// [`started_after`](Self::started_after) is the one that filters on starting.
    pub created_after: Option<Timestamp>,
    /// Created at or before this instant.
    pub created_before: Option<Timestamp>,
    /// Finished at or after this instant.
    pub completed_after: Option<Timestamp>,
    /// Finished at or before this instant.
    pub completed_before: Option<Timestamp>,
    /// Started at or after this instant.
    ///
    /// Not `dequeued_after`, although the column is written when a queued workflow is dequeued. It
    /// is not queue-specific — it is [`WorkflowRecord::started_at`], the `started_at_epoch_ms`
    /// column — and naming it after the queue implies a filter that would exclude workflows that
    /// never sat on one.
    pub started_after: Option<Timestamp>,
    /// Started at or before this instant. Not `dequeued_before`, for the same reason.
    pub started_before: Option<Timestamp>,

    /// Encoded JSON the workflow's attributes must contain.
    ///
    /// Containment, not equality: `{"tenant": "acme"}` matches a workflow with that key among
    /// others. Served by the GIN index on the column.
    pub attributes: Option<&'a str>,

    /// Most rows to return.
    pub limit: Option<i64>,
    /// Rows to skip.
    pub offset: Option<i64>,
    /// Newest first, rather than oldest first.
    pub sort_desc: bool,
    /// Whether to load the `input` column.
    ///
    /// Listing thousands of workflows pulls their arguments with them, which is usually not what
    /// the caller wanted. Turning this off leaves [`WorkflowRecord::input`] `None` — which is
    /// indistinguishable from a workflow that has no input.
    pub load_input: bool,
    /// Whether to load the `output` and `error` columns, with the same caveat.
    pub load_output: bool,
}

impl Default for WorkflowFilter<'_> {
    /// Narrows nothing, and loads everything.
    ///
    /// `load_input` and `load_output` default *on*, so a caller that does not think about them gets
    /// whole records rather than silently empty payloads. That is the one place this type cannot be
    /// `#[derive(Default)]`.
    fn default() -> Self {
        Self {
            workflow_ids: Vec::new(),
            workflow_id_prefixes: Vec::new(),
            app_scope: AppScope::Unset,
            names: Vec::new(),
            class_names: Vec::new(),
            config_names: Vec::new(),
            status: Vec::new(),
            app_versions: Vec::new(),
            executor_ids: Vec::new(),
            authenticated_users: Vec::new(),
            queue_names: Vec::new(),
            queues_only: false,
            schedule_names: Vec::new(),
            deduplication_ids: Vec::new(),
            is_debounced: None,
            parent_workflow_ids: Vec::new(),
            has_parent: None,
            forked_from: Vec::new(),
            is_fork: None,
            was_forked_from: None,
            created_after: None,
            created_before: None,
            completed_after: None,
            completed_before: None,
            started_after: None,
            started_before: None,
            attributes: None,
            limit: None,
            offset: None,
            sort_desc: false,
            load_input: true,
            load_output: true,
        }
    }
}

/// A step's recorded result, as `operation_outputs` holds it.
///
/// One row per `(workflow_id, step_id)` — that pair is the primary key, and is what makes a
/// replayed workflow skip work it has already done.
///
/// **The field names do not match the column names.** `step_id` and `step_name` are stored in
/// `function_id` and `function_name`, which is what the schema has called them since migration 1
/// and what every SDK sharing the database still reads and writes. "Function" is what these were
/// called before steps had a name of their own — the columns cannot be renamed without a migration
/// every SDK must agree on, but the Rust API need not inherit the old word.
///
/// Serde-able for the same reason as [`WorkflowRecord`], and `#[serde(default)]` per field for
/// the same reason too: a step listing taken from inside a workflow is itself a checkpointed
/// step, replays from what it recorded, and the build that reads the record is rarely the one
/// that wrote it. The exemptions here are the primary key and the name a replay matches on —
/// defaulting those would let a record this build cannot fully read pass as step 0 of a step
/// with no name, which is worse than reporting it.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct StepRecord {
    /// The workflow the step belongs to.
    pub workflow_id: String,
    /// The step's position in the workflow, counted from zero. Column `function_id`.
    ///
    /// `i32`, matching `INT4`. The count is per workflow execution, not global.
    pub step_id: i32,
    /// The registered step name, checked on replay. Column `function_name`.
    pub step_name: String,
    /// Encoded return value, if the step returned one.
    #[serde(default)]
    pub output: Option<String>,
    /// Encoded error, if the step raised one. Never set alongside `output`.
    #[serde(default)]
    pub error: Option<String>,
    /// The workflow this step started, for steps that are child-workflow calls.
    #[serde(default)]
    pub child_workflow_id: Option<String>,
    /// How `output` and `error` are encoded.
    #[serde(default)]
    pub serialization: Option<String>,
    /// When the step began.
    #[serde(default)]
    pub started_at: Option<Timestamp>,
    /// When the step finished.
    ///
    /// Also the tie-breaker on a duplicate record: a second write carrying a *different*
    /// completion time is another executor, while one carrying the same is this caller's own
    /// retry. See [`SystemDatabase::record_step`](crate::sysdb::SystemDatabase::record_step).
    #[serde(default)]
    pub completed_at: Option<Timestamp>,
}

/// How a step or a workflow ended: with a value, or with an error.
///
/// One type for both, because they end the same way and are stored the same way — an `output`
/// column and an `error` column, on `operation_outputs` and `workflow_status` respectively. The
/// variants are named after those columns.
///
/// Those two columns are independently nullable, so a row could carry both — which would make
/// the work simultaneously successful and failed, and leave a replay believing whichever the
/// reader happened to check first. A sum type means there is nothing to assert.
///
/// For a workflow this also settles the status, which is why
/// [`record_workflow_outcome`](crate::sysdb::SystemDatabase::record_workflow_outcome) does not
/// take one: an output means `SUCCESS` and an error means `ERROR`, with no third choice to get
/// wrong.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Outcome<'a> {
    /// The work returned.
    ///
    /// `None` is a void return, which is a *success* and not an absent result. Whether a step ran
    /// at all is answered by the presence of its row, never by this being empty.
    Output(Option<&'a str>),
    /// The work raised, carrying the encoded error.
    Error(&'a str),
}

impl<'a> Outcome<'a> {
    /// The terminal status this outcome puts a workflow in.
    pub fn status(self) -> WorkflowStatus {
        match self {
            Outcome::Output(_) => WorkflowStatus::Success,
            Outcome::Error(_) => WorkflowStatus::Error,
        }
    }

    /// The pair of column values, in the order the tables hold them.
    pub(crate) fn columns(self) -> (Option<&'a str>, Option<&'a str>) {
        match self {
            Outcome::Output(value) => (value, None),
            Outcome::Error(message) => (None, Some(message)),
        }
    }
}

/// How somebody else's workflow ended.
///
/// The read counterpart of [`Outcome`], and it has four variants where that has two. `Outcome` is
/// what a *run* reports, and a run can only return or raise. Cancelling and parking are done to a
/// workflow rather than by it, so they are states no run ever reports — and it is precisely the
/// caller waiting on a workflow it is not running that has to be told about them.
///
/// **"Awaited" is load-bearing rather than decorative:** a cancelled *awaited* workflow must not
/// be mistaken for the *awaiting* one having been cancelled. The same distinction is why the
/// variants below are values — see the second bullet.
///
/// Not named for terminality, because one variant is not terminal:
/// [`WorkflowStatus::is_terminal`] is deliberately false for
/// [`MaxRecoveryAttemptsExceeded`](WorkflowStatus::MaxRecoveryAttemptsExceeded), which
/// [`Parked`](Self::Parked) is. It ends a *wait* without ending the workflow, since the workflow can
/// still be resumed.
///
/// **All four come back as values, including the two that are failures.** Two reasons, and the
/// second is the one that would be a bug:
///
/// - A failed workflow's error is encoded in whatever the workflow chose, and this layer does not
///   deserialize payloads. So a failure is already data here rather than something to raise.
/// - **A cancelled workflow must not be reported as the caller being cancelled.**
///   [`Error::WorkflowCancelled`] means the opposite thing
///   — the step-replay check refusing to run *this* workflow's steps — and returning it here would
///   read as the waiter having been cancelled. Reporting all four uniformly leaves the
///   user-facing error taxonomy to the engine, which owns it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AwaitedOutcome {
    /// It returned. `None` is a void return, which is a success and not an absent result.
    Succeeded {
        /// Encoded return value.
        output: Option<String>,
        /// How `output` is encoded.
        serialization: Option<String>,
    },
    /// It raised, carrying the encoded error.
    Failed {
        /// Encoded error.
        error: String,
        /// How `error` is encoded.
        serialization: Option<String>,
    },
    /// It was cancelled, so it has neither a value nor an error of its own.
    Cancelled,
    /// It was recovered too many times and parked, so it has neither.
    ///
    /// The limit it exceeded is not stored — it is the engine's, applied when a dequeue dispatches
    /// the workflow, rather than a column — so the count is what can be reported.
    Parked {
        /// Recovery attempts recorded against the workflow.
        recovery_attempts: i64,
    },
}

/// When a step ran, start and finish together.
///
/// A pair rather than two fields because the database records completed steps: a start with no
/// finish is not a state this table has, and half a pair yields a duration nobody can compute.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct StepTiming {
    /// When the step began.
    pub started_at: Timestamp,
    /// When the step finished, and the token that makes recording it idempotent.
    ///
    /// Fixed before the write and **it must not move between retries of the same logical
    /// record**. On a duplicate insert the stored value is compared against this one: equal means
    /// this caller's own write whose acknowledgement was lost, different means this execution
    /// recorded the step already, which is nondeterminism. Re-stamping the clock per attempt
    /// would turn every lost acknowledgement into a spurious conflict — so hold the `StepTiming`
    /// in a variable rather than building it at the call site inside a retry loop.
    ///
    /// **A step whose every attempt checks for it before writing is the exception**, and the
    /// internal step runners are the ones that do: a record acknowledged to nobody — a success
    /// committed with the work, or a failure recorded after the work rolled back — is caught by
    /// the check on the next attempt and replayed, so it never reaches the insert. Nothing that
    /// reaches the comparison there is a lost acknowledgement, so a completion that differs from
    /// this attempt's is the right answer rather than a spurious one — which is why those paths
    /// stamp the clock after their work rather than before it.
    ///
    /// Omitting the timing altogether gives up that detection: with no recorded completion there
    /// is nothing to compare, so a duplicate write is accepted rather than reported.
    pub completed_at: Timestamp,
}

/// Rejects attributes that are not a JSON *object*.
///
/// The contract is an object, not arbitrary JSON, and it is shared with the other SDKs that read
/// the column. Since this layer takes the encoded form, so a host across an FFI boundary can hand
/// over bytes it already has, the same check has to happen here rather than falling out of the
/// type.
///
/// It matters beyond tidiness: `attributes @> …` in [`WorkflowFilter`] is containment against an
/// object, so a stored array or scalar would silently never match.
pub(crate) fn validate_attributes(attributes: Option<&str>) -> Result<(), Error> {
    let Some(json) = attributes else {
        return Ok(());
    };
    serde_json::from_str::<serde_json::Map<String, serde_json::Value>>(json).map_err(|e| {
        Error::InvalidInput {
            field: "attributes".into(),
            detail: format!("must be a JSON object: {e}"),
        }
    })?;
    Ok(())
}

/// What an attribute update does to a workflow's attributes. Each carries an encoded JSON
/// object, which the system database checks is one before writing.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AttributeUpdate<'a> {
    /// Replace them with this object, or clear them with `None`.
    Replace(Option<&'a str>),
    /// Merge this object's top-level keys into them, in one statement, so two merges of different
    /// keys never lose each other's.
    ///
    /// A key already there takes the new value; a key set to `null` is stored as `null` rather
    /// than removed. Attributes that are absent, or are not an object, count as an empty one.
    Merge(&'a str),
}

impl AttributeUpdate<'_> {
    /// Rejects an update whose attributes are not a JSON object, as
    /// [`validate_attributes`] does.
    pub(crate) fn validate(&self) -> Result<(), Error> {
        match *self {
            AttributeUpdate::Replace(attributes) => validate_attributes(attributes),
            AttributeUpdate::Merge(attributes) => validate_attributes(Some(attributes)),
        }
    }
}

/// When a delayed workflow should become eligible to run.
///
/// A sum type because the two forms are alternatives, not options: giving both would be a
/// contradiction to refuse at runtime. Both resolve to the same column.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WorkflowDelay {
    /// Wait this long from now.
    ///
    /// Resolved by the write, against the database clock, for the same reason
    /// [`NewWorkflow::delay`] is a duration: the release is decided on that clock, and
    /// `now + delay` computed by a caller would carry the caller's skew into the row. An instant
    /// past the largest one storable saturates to it, which is a delay that never elapses.
    For(Duration),
    /// Wait until this instant, written as given.
    Until(Timestamp),
}

/// The result of trying to record a workflow's final outcome.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OutcomeWrite {
    /// This process recorded the outcome.
    Recorded,
    /// Another execution got there first, and the row is already terminal, or this execution no
    /// longer owns the workflow.
    ///
    /// Not an error: the caller has lost a race it was allowed to lose, and should adopt the
    /// recorded outcome rather than overwrite it.
    AlreadyFinished,
}

/// A reading of the system database's clock, carried forward by this process's monotonic clock.
///
/// Workflow deadlines are set and compared on the database's clock, which every executor sharing
/// the database shares. A wait for one is measured from a reading of that clock on
/// [`Instant`], which no setting of this host's wall clock moves, so a host whose clock is far off
/// the database's waits exactly as long as one whose clock agrees.
///
/// The reading is the database's clock as the statement that returns it runs, paired with the
/// moment its row arrives here, so it misses the trip back: a wait measured from it ends that
/// much late, never early.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct DatabaseTime {
    epoch_ms: i64,
    read_at: Instant,
}

impl DatabaseTime {
    /// A reading of `epoch_ms`, taken now.
    ///
    /// Stamped on the clock this crate's timers run on, so the instants [`reaches`](Self::reaches)
    /// gives are ones a wait through them can be measured against: on tokio, tokio's clock, which
    /// a test can pause and advance apart from the real one.
    pub fn read(epoch_ms: i64) -> Self {
        Self {
            epoch_ms,
            read_at: crate::rt::now(),
        }
    }

    /// When, on this process's monotonic clock, the database's clock reaches `instant`: the
    /// moment of the reading for an instant already behind it.
    pub fn reaches(self, instant: Timestamp) -> Instant {
        let ahead = instant.as_epoch_ms().saturating_sub(self.epoch_ms);
        self.read_at + Duration::from_millis(u64::try_from(ahead).unwrap_or(0))
    }

    /// The database's clock now, as this reading carries it forward: the reading plus however long
    /// this process's monotonic clock has run since. Behind the database's own by the trip back,
    /// as a wait measured from the reading is.
    pub fn now(self) -> Timestamp {
        let elapsed = crate::rt::now().saturating_duration_since(self.read_at);
        let elapsed = i64::try_from(elapsed.as_millis()).unwrap_or(i64::MAX);
        Timestamp::from_epoch_ms(self.epoch_ms.saturating_add(elapsed))
    }
}

/// A recorded wake time, with the reading of the database's clock taken as it was recorded or
/// read back.
///
/// What [`record_sleep`](crate::sysdb::SystemDatabase::record_sleep) answers with, on a first run
/// and on a replay alike. The wake time is an instant on the database's clock, which is the one
/// clock every executor that may replay the step shares; the reading is what lets this process
/// wait for it on its own monotonic clock, so a host whose wall clock is off the database's wakes
/// on time.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WakeTime {
    /// The recorded instant, on the database's clock.
    pub at: Timestamp,
    /// The database's clock, read in the transaction that recorded or read back `at`.
    pub clock: DatabaseTime,
}

impl WakeTime {
    /// When, on this process's monotonic clock, the wake time arrives: at once for one already
    /// past.
    pub fn arrives(self) -> Instant {
        self.clock.reaches(self.at)
    }
}

/// What the database said about a workflow after initialising it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WorkflowInitResult {
    /// The stored status, which is the existing one when the row was already there.
    pub status: WorkflowStatus,
    /// Recovery attempts recorded against the workflow.
    pub recovery_attempts: i64,
    /// The workflow's absolute expiry, as the database now holds it.
    ///
    /// Reported back because it may not be the one offered: a timeout is turned into a deadline
    /// against the database's clock, and an existing row keeps the deadline it already had.
    pub deadline: Option<Timestamp>,
    /// The database's clock as the same statement saw it, which is what a wait for `deadline` is
    /// measured from.
    pub clock: DatabaseTime,
    /// The serialization format actually stored.
    ///
    /// May differ from what the caller offered: the first writer decides the format, and every
    /// later attempt has to read the payloads that are actually there.
    pub serialization: Option<String>,
    /// Whether this caller should go on to run the workflow: whether it created the row, by this
    /// call or by an earlier attempt of it whose acknowledgement was lost.
    ///
    /// `false` means another caller created it, so the row is recorded but running it would be a
    /// second execution. The caller joins it instead, whatever has become of the row.
    pub should_execute: bool,
    /// This call's creator token, which the row holds as `creator_xid` when `should_execute`.
    ///
    /// A row inserted `PENDING` is owned by its creator from the start, so this is also the
    /// `owner_xid` the execution that runs it holds, and presents on every write it makes.
    pub creator_xid: String,
}

/// Whether a name is usable as an application name.
///
/// Three to 256 characters of lowercase letters, digits, dashes and underscores — a rule shared
/// with the other SDKs on the same database, so a name registered by one is accepted by all. The
/// length is in bytes, which is the same as characters here because nothing outside ASCII passes.
pub fn is_valid_application_name(name: &str) -> bool {
    (3..=256).contains(&name.len())
        && name
            .bytes()
            .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-' || b == b'_')
}

/// Which rows a rename moves.
///
/// A sum type rather than a name plus an `adopt_unclaimed` flag. That pair admits four states and
/// only three are valid requests: naming no application and not adopting unclaimed selects no rows
/// at all, a request that could only be refused at runtime.
///
/// **Unclaimed rows are never implied**, unlike everywhere else in this feature, where
/// `IS NULL` rides along with every ownership predicate. An unclaimed row belongs to every
/// application, so taking it away from all of them is a decision rather than a consequence.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RenameFrom<'a> {
    /// Only what this application already holds.
    Application(&'a str),
    /// What this application holds, and the unclaimed rows with it.
    ApplicationAndUnclaimed(&'a str),
    /// Only the unclaimed rows — adopting work that no application owns.
    Unclaimed,
}

impl<'a> RenameFrom<'a> {
    /// The application being renamed, if the source names one.
    pub fn application(self) -> Option<&'a str> {
        match self {
            RenameFrom::Application(name) | RenameFrom::ApplicationAndUnclaimed(name) => Some(name),
            RenameFrom::Unclaimed => None,
        }
    }
}

/// The batch size a rename defaults to.
pub const DEFAULT_RENAME_BATCH_SIZE: u32 = 10_000;

/// How a rename moves the rows that do not have to move atomically.
///
/// Only the terminal workflows and their steps are batched. Those can be an application's whole
/// history, and they scope observability and garbage collection alone, so they may lag the rest of
/// the rename without anything reading a half-renamed state.
///
/// An enum rather than an `Option<u32>`, where `None` could read as either *unbatched* or
/// *unset, so the default*.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RenameBatching {
    /// One statement per table, however many rows match. Simplest, and fine for a short history.
    Unbatched,
    /// At most this many distinct workflow ids per statement.
    Batched(u32),
}

impl Default for RenameBatching {
    fn default() -> Self {
        RenameBatching::Batched(DEFAULT_RENAME_BATCH_SIZE)
    }
}

/// What a rename moved, by table.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct ApplicationRowCounts {
    /// Queue registrations.
    pub queues: u64,
    /// Schedule registrations.
    pub schedules: u64,
    /// Registered application versions.
    pub versions: u64,
    /// Workflows, in-flight and terminal together.
    pub workflows: u64,
    /// Recorded steps.
    pub steps: u64,
}

/// How long a system-database connection may sit idle inside an open transaction before the
/// server ends its session, as `idle_in_transaction_session_timeout`.
///
/// A process that stalls mid-transaction (a frozen container, a network partition) would otherwise
/// hold the transaction's row locks, and the connection, for as long as the stall lasts.
///
/// **Connection options that already set the parameter keep their own** (the database URL's
/// `options` parameter, or `PGOPTIONS`), whatever this says. Otherwise [`Default`](Self::Default)
/// and [`After`](Self::After) replace the server's setting, one made with `ALTER ROLE` or
/// `ALTER DATABASE` included, and [`ServerSetting`](Self::ServerSetting) leaves it alone.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum IdleTransactionTimeout {
    /// [`IdleTransactionTimeout::DEFAULT`], sixty seconds.
    #[default]
    Default,
    /// Whatever the server is set to, which may be no timeout at all.
    ServerSetting,
    /// After this long: more than zero, and at most [`IdleTransactionTimeout::MAX`]. A positive
    /// value under a millisecond is a millisecond, the server's unit.
    After(Duration),
}

impl IdleTransactionTimeout {
    /// What [`Default`](Self::Default) is.
    pub const DEFAULT: Duration = Duration::from_secs(60);

    /// The longest timeout the server takes: the parameter is a 32-bit count of milliseconds,
    /// about 24.8 days.
    pub const MAX: Duration = Duration::from_millis(i32::MAX as u64);

    /// Rejects an [`After`](Self::After) the server can't take: zero, or longer than
    /// [`MAX`](Self::MAX).
    pub(crate) fn validate(&self) -> Result<(), Error> {
        let detail = match self {
            IdleTransactionTimeout::After(timeout) if timeout.is_zero() => {
                "must be more than zero; `ServerSetting` leaves the server's setting".to_owned()
            }
            IdleTransactionTimeout::After(timeout) if *timeout > Self::MAX => {
                format!("must be at most {:?}", Self::MAX)
            }
            _ => return Ok(()),
        };
        Err(Error::InvalidInput {
            field: "idle_transaction_timeout".into(),
            detail,
        })
    }

    /// The timeout to set on a connection, in the server's milliseconds, or `None` to leave the
    /// server's setting.
    pub(crate) fn millis(self) -> Option<u128> {
        match self {
            IdleTransactionTimeout::Default => Some(Self::DEFAULT.as_millis()),
            IdleTransactionTimeout::ServerSetting => None,
            IdleTransactionTimeout::After(timeout) => Some(timeout.as_millis().max(1)),
        }
    }
}

/// How many polling reads may run at once against a system database's pool.
///
/// Every wait is a loop that re-queries the database, and each pass takes a connection. Uncapped,
/// enough waiters empty the pool and starve the control plane — enqueue, dequeue, status writes,
/// recovery, cancellation — leaving the waiters blocked on writes that can no longer happen.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum PollingConcurrency {
    /// Half the pool, and at least one. The half is what leaves the other half for the control
    /// plane, and the minimum matters at `max_connections = 1`, where half is zero: a cap of zero
    /// would not be a small budget but a permanent block.
    #[default]
    Default,
    /// No cap.
    Unlimited,
    /// At most this many. Above the pool size is the caller's business: this caps polling, the
    /// pool caps connections.
    AtMost(std::num::NonZeroU32),
}

impl PollingConcurrency {
    /// The permits a cap of this size takes, for a pool of `pool_size` connections.
    ///
    /// No cap is a permit count nothing will exhaust rather than an absent semaphore, so a handle
    /// holds one unconditional [`Semaphore`](tokio::sync::Semaphore) with no branch on the acquire
    /// path.
    pub(crate) fn permits(self, pool_size: u32) -> usize {
        match self {
            PollingConcurrency::Default => usize::max((pool_size / 2) as usize, 1),
            PollingConcurrency::Unlimited => tokio::sync::Semaphore::MAX_PERMITS,
            PollingConcurrency::AtMost(n) => n.get() as usize,
        }
    }
}

/// How many workflows a queue may start per window.
///
/// One value rather than two `Option`s, because the two columns behind it are only meaningful
/// together: a limit with no window and a window with no limit are both unenforceable.
///
/// The schema stores the pair as two nullable columns and cannot enforce that, so a row with one
/// set and not the other — a state this type says cannot exist — is read as no limit. See
/// `queue_from_row`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RateLimit {
    /// Workflows that may start in each window.
    pub limit: i32,
    /// The window the starts are counted over.
    pub period: Duration,
}

/// Whether a partial update touches a field.
///
/// The third state a plain `Option` cannot carry: for a nullable column, "leave it alone" and
/// "set it to NULL" are different requests, and `Option<T>` has one spelling for both.
///
/// [`Change::Leave`] leaves the column out of the `SET` clause, and [`Change::Set`] writes the
/// inner value — including null. They are separate constructors, so there is no third thing to
/// write: no way to pass a value that is missing rather than null.
///
/// Clearability lives in the *inner* type rather than in a third variant, so it mirrors the
/// column: `Change<Option<i32>>` can leave, clear or set, while `Change<bool>` can only leave or
/// set, because the column behind it is `NOT NULL`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Change<T> {
    /// Leave whatever is stored.
    Leave,
    /// Store this instead.
    Set(T),
}

impl<T> Default for Change<T> {
    /// Leaves the field alone, so `..Default::default()` narrows an update rather than widening
    /// it. Hand-written because deriving would demand `T: Default`, which says nothing here.
    fn default() -> Self {
        Change::Leave
    }
}

impl<T> Change<T> {
    /// The value to store, if this changes anything.
    pub fn set(self) -> Option<T> {
        match self {
            Change::Set(value) => Some(value),
            Change::Leave => None,
        }
    }

    /// Whether this leaves the field alone.
    pub fn is_leave(&self) -> bool {
        matches!(self, Change::Leave)
    }
}

/// The fields of a registered queue that a partial update may change.
///
/// Separate from [`NewQueue`] because they answer different questions: registering describes a
/// whole queue, updating names only what moves. One type for both would make an absent field mean
/// "no limit" on creation and "leave unchanged" on update — the same value meaning two things
/// depending on which call it is passed to.
///
/// **Ownership is not here.** A queue changes hands only through
/// [`SystemDatabase::rename_application`](crate::sysdb::SystemDatabase::rename_application);
/// letting an update reassign it would be a silent takeover of a peer's queue.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct QueueUpdate {
    /// See [`QueueRecord::concurrency`].
    pub concurrency: Change<Option<i32>>,
    /// See [`QueueRecord::worker_concurrency`].
    pub worker_concurrency: Change<Option<i32>>,
    /// See [`QueueRecord::rate_limit`]. One field for both columns, so an update cannot leave
    /// half a limit behind.
    pub rate_limit: Change<Option<RateLimit>>,
    /// See [`QueueRecord::partition_concurrency`].
    pub partition_concurrency: Change<Option<i32>>,
    /// See [`QueueRecord::partition_worker_concurrency`].
    pub partition_worker_concurrency: Change<Option<i32>>,
    /// See [`QueueRecord::partition_rate_limit`]. One field for both columns, so an update cannot
    /// leave half a limit behind.
    pub partition_rate_limit: Change<Option<RateLimit>>,
    /// See [`QueueRecord::polling_interval`].
    pub polling_interval: Change<Duration>,
}

/// What [`SystemDatabase::upsert_queue`](crate::sysdb::SystemDatabase::upsert_queue) returns: the
/// row the registration found, if any, and the row as the registration left it, both read in its
/// transaction.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UpsertedQueue {
    /// The row that was there before, read under the registration's own lock, or `None` if this
    /// call created the queue.
    pub before: Option<QueueRecord>,
    /// The row as the registration left it: what was written, or, under
    /// [`OnExistingQueue::Leave`], what was already stored.
    pub after: QueueRecord,
}

impl UpsertedQueue {
    /// Whether this call created the queue rather than finding it.
    pub fn created(&self) -> bool {
        self.before.is_none()
    }
}

/// What [`SystemDatabase::update_queue`](crate::sysdb::SystemDatabase::update_queue) returns: the
/// row as the update's transaction read it, and the row as it wrote it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UpdatedQueue {
    /// The row the update replaced, read under the update's own lock.
    pub before: QueueRecord,
    /// The row as written.
    pub after: QueueRecord,
}

impl QueueUpdate {
    /// This update applied to a record, giving the row as it would be after the write.
    ///
    /// What a caller's validation is handed: a limit is rarely wrong on its own and usually wrong
    /// only beside another, so the merged result is the only thing worth checking.
    pub fn apply_to(&self, record: &QueueRecord) -> QueueRecord {
        QueueRecord {
            name: record.name.clone(),
            concurrency: self.concurrency.set().unwrap_or(record.concurrency),
            worker_concurrency: self
                .worker_concurrency
                .set()
                .unwrap_or(record.worker_concurrency),
            rate_limit: self.rate_limit.set().unwrap_or(record.rate_limit),
            partition_concurrency: self
                .partition_concurrency
                .set()
                .unwrap_or(record.partition_concurrency),
            partition_worker_concurrency: self
                .partition_worker_concurrency
                .set()
                .unwrap_or(record.partition_worker_concurrency),
            partition_rate_limit: self
                .partition_rate_limit
                .set()
                .unwrap_or(record.partition_rate_limit),
            polling_interval: self
                .polling_interval
                .set()
                .unwrap_or(record.polling_interval),
            // Not updatable: the name is the queue's address, and ownership moves only by rename.
            application_name: record.application_name.clone(),
        }
    }

    /// Whether this would change nothing.
    pub fn is_empty(&self) -> bool {
        self.concurrency.is_leave()
            && self.worker_concurrency.is_leave()
            && self.rate_limit.is_leave()
            && self.partition_concurrency.is_leave()
            && self.partition_worker_concurrency.is_leave()
            && self.partition_rate_limit.is_leave()
            && self.polling_interval.is_leave()
    }
}

/// A request to debounce a workflow onto a deduplication key.
///
/// A struct rather than seven parameters because several of them are `&str` or `Option<&str>`,
/// and transposing two would compile; named fields rule that out.
///
/// **No `new` and no `Default`**, unlike [`NewWorkflow`] and [`NewQueue`], which take a single
/// identifying string. A constructor here would take several `&str` in a row and hand back the
/// transposition this type exists to prevent, and a default would fabricate a request with empty
/// names — the values the enqueue side rejects. Every field is named at the call site on purpose.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct DebounceRequest<'a> {
    /// The workflow being debounced, matched on all three parts of its name so a bounce extends
    /// only a workflow of the same name, class and configured instance.
    ///
    /// Two configured instances of one class share a name *and* a class, so without the instance
    /// a bounce for one could extend the other and replace its inputs; and a key collision between
    /// unrelated workflows never overwrites the wrong one's inputs.
    pub workflow: &'a QualifiedWorkflowName,
    /// The queue the debounced workflow sits on.
    pub queue_name: &'a str,
    /// The debounce key, held in `deduplication_id` while the workflow is delayed.
    pub deduplication_id: &'a str,
    /// How long from the bounce to release the workflow, capped at its debounce deadline.
    ///
    /// A duration the bounce resolves against the database clock, as the insert resolved the
    /// first delay and the deadline.
    pub period: Duration,
    /// The inputs to run with, replacing whatever the previous request left.
    pub inputs: Option<&'a str>,
    /// How `inputs` is encoded.
    pub serialization: Option<&'a str>,
    /// The application the bounce acts for; `None` means the writing handle's own.
    pub application_name: Option<&'a str>,
}

impl DebounceRequest<'_> {
    /// Rejects the values no writer could have stored, so a bounce cannot silently match nothing.
    ///
    /// The same rule [`NewWorkflow::validate`] applies on the way in: an empty value is absent
    /// rather than a value, and matching one would find no row and report the key unheld —
    /// indistinguishable from a key nobody holds. A [`QualifiedWorkflowName`] never holds an empty
    /// class or config. The period is held to the rule `delay` is: a positive duration that fits in
    /// milliseconds.
    pub fn validate(&self) -> Result<(), Error> {
        validate_positive_duration("period", self.period)?;
        for (field, value) in [
            ("workflow.name", self.workflow.name()),
            ("queue_name", self.queue_name),
            ("deduplication_id", self.deduplication_id),
        ] {
            if value.is_empty() {
                return Err(Error::InvalidInput {
                    field: field.into(),
                    detail: "must not be empty".to_owned(),
                });
            }
        }
        for (field, value) in [
            ("inputs", self.inputs),
            ("serialization", self.serialization),
            ("application_name", self.application_name),
        ] {
            if value == Some("") {
                return Err(Error::InvalidInput {
                    field: field.into(),
                    detail: "must be absent rather than empty".to_owned(),
                });
            }
        }
        Ok(())
    }
}

/// What a debounce did, or why it could not.
///
/// Recorded as a debounce step's output when one runs inside a workflow, so a replay reports what
/// the first run did rather than bouncing again — see
/// [`SystemDatabase::debounce_delayed_workflow`](crate::sysdb::SystemDatabase::debounce_delayed_workflow).
///
/// Three outcomes rather than a flat record carrying a bounced workflow's id alongside the
/// holder's fields, of which exactly one group would ever be populated. Reading such a record
/// means checking which — and a stored one means *guessing* which, since nothing in the shape
/// rules out both groups being set at once. Named states cannot encode that, so the ambiguity is
/// gone rather than resolved by convention.
///
/// **The serialized form is this crate's own, and it is this type's derive.** A step is only ever
/// replayed by the SDK that wrote it — workflows cross languages by enqueue alone — so the encoding
/// is recorded in the `serialization` column rather than agreed between SDKs: there is no common
/// wire form to conform to. The consequence is that renaming a variant or field here changes what a
/// replay expects to read, so a rename wants the same care a schema change does.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum Debounce {
    /// The existing delayed workflow was extended, and its inputs replaced.
    Bounced {
        /// The workflow whose delay moved.
        workflow_id: String,
    },
    /// The key is held by a workflow this bounce could not extend, described so the caller can
    /// tell a legitimate collision from a coincidence.
    Held(DebounceHolder),
    /// Nothing holds the key, so the caller should start a fresh workflow.
    Unheld,
}

/// The workflow holding a deduplication key, when a debounce could not extend it.
///
/// A bounce fails for more than one reason and they want different responses, so the holder is
/// described rather than merely reported: a **different name, class or instance** under the same
/// key is a collision between unrelated workflows (`"a" + "b-c"` against `"a-b" + "c"`, or two
/// configured instances of one class), a holder that is **not debounced** is an ordinary
/// deduplicated enqueue, and a **different application** means the collision is across
/// applications sharing the database.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct DebounceHolder {
    /// The workflow holding the key.
    pub workflow_id: String,
    /// Whether the holder is itself a debounced workflow.
    pub is_debounced: bool,
    /// The holder's registered function name.
    pub workflow_name: Option<String>,
    /// The holder's class, for a class-bound workflow.
    pub class_name: Option<String>,
    /// The holder's configured instance name, for an instance-bound workflow.
    pub config_name: Option<String>,
    /// The application that owns the holder, or `None` if it is unclaimed.
    pub application_name: Option<String>,
}

/// A queue as the registry holds it.
///
/// The registry exists so a queue's limits can be read and changed without the executor that
/// declared it — a running deployment is not the only thing that knows what a queue is.
///
/// Both periods are `DOUBLE PRECISION` seconds in the schema rather than the integer
/// milliseconds used elsewhere, which is the divergence the module documentation warns about;
/// they are read through [`duration_from_secs`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct QueueRecord {
    /// The queue's name, which is its address across every application sharing the database.
    pub name: String,
    /// Workflows this queue may have running at once, across all executors. `None` is unlimited.
    pub concurrency: Option<i32>,
    /// Workflows one executor may have running at once. `None` is unlimited.
    pub worker_concurrency: Option<i32>,
    /// How fast workflows may start, or `None` for unthrottled.
    pub rate_limit: Option<RateLimit>,
    /// Workflows one partition may have running at once, across all executors.
    ///
    /// Setting any of the three partition limits is what partitions a queue. Each applies within
    /// one partition rather than to the queue as a whole, so they sit beside the queue-wide
    /// limits rather than replacing them: both are enforced. A per-partition concurrency limit
    /// may not exceed its queue-wide counterpart, which it could then never bind under. The two
    /// rate limits are validated separately and never compared, because their windows may
    /// differ: a per-partition limit over a shorter window caps bursts the queue-wide one allows.
    pub partition_concurrency: Option<i32>,
    /// Workflows one executor may have running at once within one partition.
    pub partition_worker_concurrency: Option<i32>,
    /// How fast workflows may start within one partition, or `None` for unthrottled.
    pub partition_rate_limit: Option<RateLimit>,
    /// How often an idle executor asks this queue for work.
    pub polling_interval: Duration,
    /// The application that owns the queue, or `None` if it is unclaimed.
    pub application_name: Option<String>,
}

impl QueueRecord {
    /// Whether the queue is partitioned, which any per-partition limit makes it.
    ///
    /// Asked of the limits alone. The `partition_queue` column is written, derived from them, and
    /// never read: a row with the flag set and only queue-wide limits is an unpartitioned queue
    /// with queue-wide limits, and every limit applies at the scope its column names.
    pub fn is_partitioned(&self) -> bool {
        has_partition_limit(
            self.partition_concurrency,
            self.partition_worker_concurrency,
            self.partition_rate_limit,
        )
    }
}

/// Whether any per-partition limit is set, which is what partitions a queue.
///
/// The one statement of the rule, shared by the stored row, the row to register, and the engine's
/// `Queue`, since it also decides what the `partition_queue` column holds.
pub(crate) fn has_partition_limit(
    partition_concurrency: Option<i32>,
    partition_worker_concurrency: Option<i32>,
    partition_rate_limit: Option<RateLimit>,
) -> bool {
    partition_concurrency.is_some()
        || partition_worker_concurrency.is_some()
        || partition_rate_limit.is_some()
}

/// A queue to register, as the caller supplies it.
///
/// Borrowed and separate from [`QueueRecord`] for the reason [`NewWorkflow`] is separate from
/// [`WorkflowRecord`]: registering does not require an owner, and the record always reports one.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NewQueue<'a> {
    /// The queue's name.
    pub name: &'a str,
    /// See [`QueueRecord::concurrency`].
    pub concurrency: Option<i32>,
    /// See [`QueueRecord::worker_concurrency`].
    pub worker_concurrency: Option<i32>,
    /// See [`QueueRecord::rate_limit`].
    pub rate_limit: Option<RateLimit>,
    /// See [`QueueRecord::partition_concurrency`].
    pub partition_concurrency: Option<i32>,
    /// See [`QueueRecord::partition_worker_concurrency`].
    pub partition_worker_concurrency: Option<i32>,
    /// See [`QueueRecord::partition_rate_limit`].
    pub partition_rate_limit: Option<RateLimit>,
    /// See [`QueueRecord::polling_interval`].
    pub polling_interval: Duration,
    /// The application to register the queue for; `None` means the writing handle's own.
    pub application_name: Option<&'a str>,
}

impl NewQueue<'_> {
    /// Whether the queue is partitioned, which any per-partition limit makes it. The same rule as
    /// [`QueueRecord::is_partitioned`], for the row before it is written.
    pub fn is_partitioned(&self) -> bool {
        has_partition_limit(
            self.partition_concurrency,
            self.partition_worker_concurrency,
            self.partition_rate_limit,
        )
    }
}

impl<'a> NewQueue<'a> {
    /// A queue with no limits, polling once a second.
    pub fn new(name: &'a str) -> Self {
        Self {
            name,
            concurrency: None,
            worker_concurrency: None,
            rate_limit: None,
            partition_concurrency: None,
            partition_worker_concurrency: None,
            partition_rate_limit: None,
            polling_interval: Duration::from_secs(1),
            application_name: None,
        }
    }
}

/// What registering a queue that already exists should do to it.
///
/// Never to its owner, which moves only by rename: a name already held by another application is
/// [`Error::RegisteredByAnother`] in both cases, because the name is the queue's address — as long
/// as the caller has a name to be refused under. A nameless one is let through and [`Update`]
/// replaces the peer's limits: a known limit, and an open question shared with the other SDKs on
/// the same database.
///
/// [`Update`]: OnExistingQueue::Update
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OnExistingQueue {
    /// Overwrite the stored limits with the ones supplied.
    Update,
    /// Leave the stored row exactly as it is.
    Leave,
}

/// What one claim from a queue took, and why it took nothing if it did.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Dequeued {
    /// The workflows this claim started, in dequeue order.
    pub workflow_ids: Vec<String>,
    /// The limit that left this claim no room, when one did.
    ///
    /// Set only when the claim returned before selecting anything, so `workflow_ids` is then
    /// empty. `None` with nothing claimed means no row was eligible, or a peer took them first.
    pub stopped_by: Option<DequeueStop>,
}

/// The limit that left a claim no room to start anything.
///
/// Each is the first limit, in the order the claim checks them, whose remaining room came to zero:
/// worker concurrency, then the rate limits, then the concurrency limits, the queue-wide one ahead
/// of the per-partition one at each step.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DequeueStop {
    /// This process is already running as many of the queue's workflows as it may.
    WorkerConcurrency,
    /// This process is already running as many of the partition's workflows as it may.
    PartitionWorkerConcurrency,
    /// The queue's rate-limit window is full.
    RateLimit,
    /// The partition's rate-limit window is full.
    PartitionRateLimit,
    /// The queue already has as many workflows `PENDING` as its concurrency allows.
    Concurrency,
    /// The partition already has as many workflows `PENDING` as its concurrency allows.
    PartitionConcurrency,
}

impl DequeueStop {
    /// Whether the limit is counted across the whole queue, whichever partition was claimed from.
    ///
    /// Such a limit leaves every other partition just as full, so a caller walking the partitions
    /// one at a time can end the walk on it rather than paying a claim per partition to learn the
    /// same thing. A per-partition limit says nothing about the next key.
    pub fn spans_partitions(self) -> bool {
        match self {
            Self::WorkerConcurrency | Self::RateLimit | Self::Concurrency => true,
            Self::PartitionWorkerConcurrency
            | Self::PartitionRateLimit
            | Self::PartitionConcurrency => false,
        }
    }
}

/// A registered version of the application.
///
/// The registry is what lets a firing schedule stamp the *latest* version, so only executors
/// running that code dequeue it, and what [`WorkflowRecord::app_version`] refers to.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct VersionInfo {
    /// The application that registered this version, or `None` if it is unclaimed.
    pub app_name: Option<String>,
    /// Generated identity, distinct from the name.
    pub version_id: String,
    /// The version as the application names it — the value stored on workflow rows.
    pub version_name: String,
    /// Orders the registry: the latest version is the one with the highest value here, not the
    /// one created most recently.
    pub version_timestamp: Timestamp,
    /// When the row was first written.
    pub created_at: Timestamp,
}

/// A message sent to a workflow, as `notifications` holds it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NotificationRecord {
    /// The message's identity, and the table's primary key.
    ///
    /// Defaulted by the database with `gen_random_uuid()`, so a sender never supplies one. It is
    /// what a `recv` marks consumed, and the only thing distinguishing two identical messages on
    /// the same topic.
    pub message_uuid: String,
    /// The topic it was sent on, or `None` for the default topic.
    pub topic: Option<String>,
    /// Encoded payload.
    pub message: String,
    /// How `message` is encoded.
    pub serialization: Option<String>,
    /// When it was sent.
    pub created_at: Timestamp,
    /// Whether a `recv` has taken it.
    ///
    /// Receiving marks rather than deletes, so a consumed message stays visible to export and to
    /// anyone auditing what a workflow was sent.
    pub consumed: bool,
}

/// How a [`StepCaller`] encodes the result this layer records for it, and decodes it on replay.
///
/// **This layer owns the value but not its encoding.** A management call made from a workflow —
/// a listing, a cancel, a fork, a schedule change — records its own result as the caller's step,
/// in the transaction that does the work, so a replay reads back exactly what the first run saw.
/// The value is this layer's ([`WorkflowRecord`]s, [`StepRecord`]s, a [`Debounce`], an id list,
/// a recorded [`Error`]), but how it is written is the application's choice, the same one that
/// governs its own steps' outputs. The caller supplies it here, and the row's `serialization`
/// column records the tag it gives.
///
/// **The value crosses as JSON data, not as a Rust type.** This layer turns its value into a
/// [`serde_json::Value`] and hands that over, and gets one back on replay. That keeps it free of
/// any serializer machinery of its own, and leaves something a host in another language can
/// implement: JSON in, text and a tag out. Every value that crosses is a plain record — strings,
/// integers, enums, epoch-millisecond times — so nothing depends on a format being
/// human-readable.
pub trait StepCodec: Send + Sync + fmt::Debug {
    /// Encodes `value`, and names the encoding for the row's `serialization` column.
    fn encode(&self, value: serde_json::Value) -> Result<EncodedValue, Error>;

    /// Decodes `value`, stored under the tag `serialization`, back into JSON data.
    ///
    /// A tag this codec cannot read is [`Error::UnknownSerialization`], never a guess. An absent
    /// tag is the codec's to interpret.
    fn decode(&self, value: &str, serialization: Option<&str>) -> Result<serde_json::Value, Error>;
}

/// Encodes a value to record as a step's output or error, through the caller's codec. `what` names
/// the value for the failure message: `an output` or `an error`. The reverse of
/// [`decode_step_value`].
///
/// Every value this crate records through a [`StepCodec`] goes through here (a management call's
/// result, an internal step's output, a recorded system-database error), so a replay reads them all
/// back the same way. An ordinary step's output and error are the application's own types, and go
/// through the payload encoding instead.
///
/// `serde_json` fails only on a type that cannot be JSON, such as a non-string map key, and turns a
/// non-finite float into `null` without failing, so a value here must hold none; every value this
/// takes is one of the engine's own plain records or an [`Error`]. The codec can fail too, since a
/// custom serializer may fail for its own reasons.
pub(crate) fn encode_step_value<T: serde::Serialize + ?Sized>(
    value: &T,
    codec: &dyn StepCodec,
    what: &str,
) -> Result<EncodedValue, Error> {
    let value = serde_json::to_value(value)
        .map_err(|e| Error::Malformed(format!("{what} to record is not JSON: {e}")))?;
    codec.encode(value)
}

/// Reads one of `step`'s recorded values back through the caller's codec, by the tag the row was
/// stored under, and says which step it came from if it cannot. `what` names the value for that
/// message: `an error` or `an output`.
///
/// The codec and serde don't know which step they are reading, so a failure gets that here:
///
/// - [`Error::UnknownSerialization`] is returned as it is, because callers match on it and it names
///   only the tag. The step that holds it is logged instead.
/// - [`Error::Malformed`] gets the step in its message.
/// - Any other error, such as one a custom serializer's decode returns, is returned unchanged.
pub(crate) fn decode_step_value<T: serde::de::DeserializeOwned>(
    step: &StepRecord,
    recorded: &str,
    caller: StepCaller<'_>,
    step_name: &str,
    what: &str,
) -> Result<T, Error> {
    let StepCaller {
        workflow_id,
        step_id,
        codec,
        ..
    } = caller;
    codec
        .decode(recorded, step.serialization.as_deref())
        .and_then(|value| {
            serde_json::from_value(value).map_err(|e| Error::Malformed(e.to_string()))
        })
        .map_err(|error| match error {
            Error::UnknownSerialization { ref name } => {
                tracing::warn!(
                    workflow_id,
                    step_id,
                    step_name,
                    serialization = %name,
                    "a recorded step holds {what} under a serialization this process cannot read"
                );
                error
            }
            Error::Malformed(detail) => Error::Malformed(format!(
                "workflow {workflow_id} step {step_id} ({step_name}) has {what} this build cannot \
                 read: {detail}"
            )),
            other => other,
        })
}

/// The execution a write is made by: the workflow it runs, and the token it owns it by.
///
/// A write that carries one lands only while the workflow's row still holds that token, checked
/// in the write's own transaction before anything else and locking the row until it commits. Once
/// the workflow has been handed to another execution (a resume, a recovery), the write is refused
/// with [`WorkflowConflict`](crate::sysdb::Error::WorkflowConflict) and rolls back. Once it has been
/// cancelled, the write is refused with
/// [`WorkflowCancelled`](crate::sysdb::Error::WorkflowCancelled) and rolls back.
/// `None` where a method takes an `Option<WorkflowOwner>` is a caller that is not an execution, a
/// client say, and is not checked.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WorkflowOwner<'a> {
    /// The workflow the execution runs.
    pub workflow_id: &'a str,
    /// The token it owns the workflow by.
    pub owner_xid: &'a str,
}

/// The workflow step a management call runs as, and how its result is recorded.
///
/// What the methods that record their own result take, where the plain-id methods take `caller:
/// Option<(&str, i32)>`. `None` at the call site is a caller outside a workflow: nothing is
/// checked and nothing is recorded, so it needs no codec either.
///
/// Given one, the call's work and its step checkpoint **commit together**: see
/// [`fork_workflows`](crate::sysdb::SystemDatabase::fork_workflows).
#[derive(Debug, Clone, Copy)]
pub struct StepCaller<'a> {
    /// The workflow making the call.
    pub workflow_id: &'a str,
    /// The step the call is recorded under.
    pub step_id: i32,
    /// How the recorded result is encoded and decoded.
    pub codec: &'a dyn StepCodec,
    /// The token the calling execution owns its workflow by: the call and its record land only
    /// while the workflow is still that execution's.
    pub owner_xid: &'a str,
}

impl<'a> StepCaller<'a> {
    /// The execution making the call, which the call is checked against.
    pub fn owner(&self) -> WorkflowOwner<'a> {
        WorkflowOwner {
            workflow_id: self.workflow_id,
            owner_xid: self.owner_xid,
        }
    }
}

/// The workflow a [`get_event`](crate::sysdb::SystemDatabase::get_event) runs on behalf of, and the
/// steps it records against.
///
/// A struct where the plain-id methods here take `caller: Option<(&str, i32)>`, because this one
/// carries **two** step ids and they are not interchangeable: one records the read itself, the
/// other the deadline it waits until. `Some(("wf", 4, 5))` says nothing about which is which, and
/// transposing them is a mistake nothing catches until a recovery replays the wrong step under the
/// wrong name.
///
/// `None` at the call site is a caller outside a workflow: it has no steps to record and nothing
/// to replay, so it waits on the wall clock and returns whatever it found. That optionality is why
/// this is a type at all — `recv` blocks and checkpoints the same two steps, but requires its
/// caller, so it takes them as plain parameters with nothing to wrap.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct GetEventCaller<'a> {
    /// The workflow doing the reading — the caller, never the workflow being read from.
    pub workflow_id: &'a str,
    /// The step the read itself is recorded under, which is what makes it replay rather than
    /// wait a second time.
    pub step_id: i32,
    /// The step the deadline is checkpointed under, recorded as a `DBOS.sleep` the caller may
    /// abandon early rather than one it waits out.
    ///
    /// Separate from `step_id` because a recovery has to resume the *same* deadline rather than
    /// start the timeout again — a read that waited fifty of its sixty seconds and crashed has ten
    /// left.
    pub timeout_step_id: i32,
    /// The token the reading execution owns its workflow by, which both records present.
    pub owner_xid: &'a str,
}

/// The workflow an [`init_workflow`](crate::sysdb::SystemDatabase::init_workflow) creates a child
/// on behalf of, and the step the start is recorded under.
///
/// Named for its one method, as [`GetEventCaller`] is, and a struct for the same reason and then a
/// second: this carries a step *name* as well as the two ids, and the name is the child workflow's
/// own rather than a library constant, so `Some(("wf", 4, "checkout"))` reads as three unrelated
/// values. `None` is a root start — a workflow begun from outside any workflow, which has no
/// parent to record against.
///
/// **Everything the parent contributes is here, including the id the child's own row carries.**
/// A `parent_workflow_id` on [`NewWorkflow`] as well would be one fact spelled twice and two
/// chances to disagree: nothing would catch a caller naming one parent on the row and another on
/// the step. Grouping it here makes the agreement structural rather than checked — a row gets a
/// parent exactly when a start is recorded against that parent, because the same value writes
/// both — and it is where [`NewWorkflow`]'s own rule puts it, that type holding the columns a
/// caller may meaningfully set and leaving the ones a mechanism owns to the mechanism.
///
/// **Passing this is what makes the two writes one.** The child's row and the parent's record of
/// having started it commit together, so no observer and no replay ever sees a child that exists
/// with nothing pointing at it. The alternative — the caller writing the record itself, on the
/// next round trip — leaves a window in which a crash, or merely a dropped future, produces
/// exactly that: a workflow nothing started.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct InitWorkflowCaller<'a> {
    /// The workflow doing the starting: the record is written against it, and the child's row
    /// points back at it.
    pub parent_workflow_id: &'a str,
    /// The step the start occupies in the parent, claimed before this call was made.
    pub step_id: i32,
    /// What the step is called, which is the child workflow's bare name — see
    /// [`record_child_workflow`](crate::sysdb::SystemDatabase::record_child_workflow), whose
    /// column this shares.
    pub step_name: &'a str,
    /// When the start began, for the row's `started_at`. The completion is stamped by the write
    /// itself, since the step spans the launch alone.
    pub started_at: Timestamp,
    /// The token the parent's execution owns it by. The child is created, and the start
    /// recorded, only while the parent is still that execution's: a stale parent starts nothing.
    pub owner_xid: &'a str,
}

/// An encoded payload and the format it is encoded in.
///
/// What the reads return: this layer moves payloads as opaque strings and never decodes one, so
/// the format has to travel with the value for the caller to make sense of it.
/// [`get_event`](crate::sysdb::SystemDatabase::get_event) and `recv` return one, and a stream read
/// returns a run of them — in [`StreamRead`], and in [`AwaitedStream::Values`]. Per value, because
/// each stream entry carries the format it was written in.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EncodedValue {
    /// The payload, exactly as it was stored.
    pub value: String,
    /// How `value` is encoded, or `None` if whoever wrote it recorded no format.
    pub serialization: Option<String>,
}

/// A key/value a workflow published, as `workflow_events` holds it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EventRecord {
    /// The key it was published under.
    pub key: String,
    /// Encoded value.
    pub value: String,
    /// How `value` is encoded.
    pub serialization: Option<String>,
}

/// A run of a stream's values from one offset, read together with its producer's liveness.
///
/// The pair is the point. A reader deciding whether to wait needs to know both whether a value is
/// there and whether the workflow that would write one is still running, and it needs them to agree
/// — read separately, a reader can find nothing at the offset, then find the workflow finished, and
/// stop one value short of a stream that was complete all along.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StreamRead {
    /// The producing workflow's status, from the same snapshot as `values`.
    ///
    /// **A terminal status does not mean the stream is finished.** Cancelling a workflow and
    /// timing one out both set the status from outside while it is still writing, so a reader that
    /// stops here must first drain to the first empty offset.
    pub status: WorkflowStatus,
    /// The values at the offset read from and the ones after it, in order and with no gaps:
    /// `values[i]` is the value at `offset + i`. Empty if nothing is written at the offset yet.
    ///
    /// The closing sentinel [`STREAM_CLOSED`](crate::sysdb::STREAM_CLOSED) arrives here like any
    /// other value; recognising it belongs to the loop, which is the only thing that knows the
    /// stream is being read rather than inspected.
    pub values: Vec<EncodedValue>,
}

/// What a blocking stream read found:
/// [`await_stream_values`](crate::sysdb::SystemDatabase::await_stream_values).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AwaitedStream {
    /// At least one value, as in [`StreamRead::values`]: the first is the one at the offset asked
    /// for, and the rest follow it with no gaps.
    Values(Vec<EncodedValue>),
    /// Nothing at the offset, and the producer is no longer running: a last read after seeing that
    /// found the offset still empty.
    ///
    /// Not a promise that nothing will follow. A producer that is cancelled or parked can be resumed
    /// and write again; this is the end of what a reader waiting now will see.
    ///
    /// A *closed* stream is not reported this way. Its sentinel is a value, and arrives in
    /// [`Values`](Self::Values).
    Ended,
    /// The deadline passed with nothing at the offset and the producer still running.
    TimedOut,
}

/// One entry of a workflow's stream, as `streams` holds it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StreamRecord {
    /// The stream this entry belongs to.
    pub key: String,
    /// Position within that stream, counted from zero.
    pub offset: i32,
    /// Encoded value.
    pub value: String,
    /// How `value` is encoded.
    pub serialization: Option<String>,
    /// The step that wrote the entry. Column `function_id`.
    ///
    /// Added by migration 6 alongside `workflow_events_history`, "to enable tracking event
    /// history by step ID and copying events during workflow forking" — so a fork can carry the
    /// entries written before its start step and leave the rest behind.
    pub step_id: i32,
}

/// One workflow to fork, and where its fork picks up.
///
/// A struct per fork rather than three parallel lists, so the original, its fork's id and its
/// start step cannot disagree in length.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Fork<'a> {
    /// The workflow to fork from. It is not modified beyond being marked as forked from.
    pub original_id: &'a str,
    /// The id the fork gets, or `None` to have one generated.
    ///
    /// The generated ids come back from
    /// [`SystemDatabase::fork_workflows`](crate::sysdb::SystemDatabase::fork_workflows).
    pub forked_id: Option<&'a str>,
    /// The first step the fork will run.
    ///
    /// Every step *below* this is copied to the fork, so it replays them instead of running them.
    /// Steps are numbered from zero, so `0` copies nothing and restarts the workflow from the
    /// beginning, and `1` carries step 0 across.
    pub start_step: i32,
}

impl<'a> Fork<'a> {
    /// A fork of `original_id` restarting from the beginning, with a generated id.
    pub fn new(original_id: &'a str) -> Self {
        Self {
            original_id,
            forked_id: None,
            start_step: 0,
        }
    }

    /// Rejects ids the schema cannot key on, matching [`NewWorkflow::validate`].
    pub(crate) fn validate(&self) -> Result<(), Error> {
        if self.original_id.is_empty() {
            return Err(Error::InvalidInput {
                field: "original_id".into(),
                detail: "must not be empty".to_owned(),
            });
        }
        // `None` asks for one to be generated; `Some("")` is an id that cannot be looked up.
        if self.forked_id == Some("") {
            return Err(Error::InvalidInput {
                field: "forked_id".into(),
                detail: "must be absent rather than empty".to_owned(),
            });
        }
        if self.start_step < 0 {
            return Err(Error::InvalidInput {
                field: "start_step".into(),
                detail: "must not be negative".to_owned(),
            });
        }
        Ok(())
    }
}

/// Which step a fork restarts from, when the caller wants it worked out rather than stated.
///
/// Taken by [`fork_from`](crate::sysdb::SystemDatabase::fork_from), which is kept for a handler of
/// Conductor's `fork_from_failure` message; the engine's own fork always states its step.
///
/// A sum type because the four are alternatives, not options: as four independent flags, every
/// combination but exactly one set would have to be refused at runtime. The resolved value is a
/// [`Fork::start_step`], so the named step *re-runs* and everything below it replays — forking
/// "from the failure" means running the failed step again, not skipping it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ForkPoint<'a> {
    /// The step that failed, or the last recorded step if none did.
    ///
    /// The fallback matters: a workflow can be left unfinished without any step recording an
    /// error — a process killed mid-step records nothing at all — and forking it should still
    /// resume where it stopped.
    LastFailure,
    /// The last recorded step, failed or not.
    LastStep,
    /// A step chosen by the caller, with no lookup.
    Step(i32),
    /// The last step recorded under this name.
    ///
    /// For a workflow whose shape is known: fork every one of them from `charge_card`, whatever
    /// position it happens to occupy in each.
    StepNamed(&'a str),
}

/// How forked workflows are created.
///
/// The defaults are what happens when the caller says nothing: the fork is enqueued on the internal
/// queue, inherits its source's version, and carries no timeout.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct ForkOptions<'a> {
    /// The version the fork runs under. `None` inherits the source's.
    ///
    /// Forking onto a *new* version is the point of the parameter: a workflow that failed on a
    /// broken deployment is forked onto the fixed one.
    pub application_version: Option<&'a str>,
    /// The queue the fork is enqueued on. `None` means [`INTERNAL_QUEUE`](crate::sysdb::INTERNAL_QUEUE).
    ///
    /// A fork is always enqueued rather than started: it is created by whoever asked for the
    /// fork, and run by whichever executor picks it up.
    pub queue_name: Option<&'a str>,
    /// The partition of that queue.
    pub queue_partition_key: Option<&'a str>,
    /// How long the fork may run before it is cancelled.
    pub timeout: Option<Duration>,
    /// Child workflow ids to rewrite as the fork's steps are copied.
    ///
    /// When a tree of workflows is forked together, the parent's recorded children are the
    /// *original* children — replaying them would make the fork adopt the originals rather than
    /// its own. Each `(from, to)` pair rewrites one `child_workflow_id` during the copy.
    pub replacement_children: &'a [(&'a str, &'a str)],
}

impl ForkOptions<'_> {
    /// Applies the same rules [`NewWorkflow::validate`] applies to the same columns.
    ///
    /// Worth stating why an empty string is not simply treated as absent: the columns here are
    /// `COALESCE`d against the source's, so `Some("")` would win over the value it was meant to
    /// leave alone — a fork stamped with an empty version rather than its source's, and so
    /// invisible to the recovery that scopes by version.
    pub(crate) fn validate(&self) -> Result<(), Error> {
        for (field, value) in [
            ("application_version", &self.application_version),
            ("queue_name", &self.queue_name),
            ("queue_partition_key", &self.queue_partition_key),
        ] {
            if *value == Some("") {
                return Err(Error::InvalidInput {
                    field: field.into(),
                    detail: "must be absent rather than empty".to_owned(),
                });
            }
        }
        // As on `NewWorkflow`: a zero timeout would expire the fork before it ran.
        if let Some(timeout) = self.timeout {
            validate_positive_duration("timeout", timeout)?;
        }
        // The replacements are applied by joining against them, so one child named twice would
        // match a copied step twice and duplicate it. The primary key would catch that as a
        // constraint violation, which says nothing about the map that caused it. Nor is taking
        // the first match an answer: two replacements for one child is a caller that does not
        // know which it wants, and picking for them is worse than saying so.
        for (index, (original, replacement)) in self.replacement_children.iter().enumerate() {
            if original.is_empty() {
                return Err(Error::InvalidInput {
                    field: "replacement_children".into(),
                    detail: "a replaced child id must not be empty".to_owned(),
                });
            }
            // A fork rewritten to wait on `""` waits on a workflow that can never exist.
            if replacement.is_empty() {
                return Err(Error::InvalidInput {
                    field: "replacement_children".into(),
                    detail: format!("the replacement for {original} must not be empty"),
                });
            }
            if self.replacement_children[..index]
                .iter()
                .any(|(earlier, _)| earlier == original)
            {
                return Err(Error::InvalidInput {
                    field: "replacement_children".into(),
                    detail: format!("{original} is replaced more than once"),
                });
            }
        }
        Ok(())
    }

    /// The timeout in milliseconds, the unit its column stores.
    ///
    /// Refused rather than saturated, by [`duration_to_ms`]. [`validate`](Self::validate) is where
    /// a caller hears it, through [`validate_positive_duration`]; the insert converts through here,
    /// so the unit is written once.
    pub(crate) fn timeout_ms(&self) -> Result<Option<i64>, Error> {
        self.timeout
            .map(|timeout| duration_to_ms("timeout", timeout))
            .transpose()
    }
}

/// Where a rewound workflow is re-enqueued.
///
/// **`None` does not mean the same thing across these fields.** The version is left as it was,
/// while the queue and its partition are the rewind's own: `None` there is the internal queue and
/// no partition, not the workflow's previous ones, so a partition key from the queue it ran on
/// does not follow it onto a queue that may not be partitioned.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct RewindOptions<'a> {
    /// The version the workflow is re-run under. `None` keeps its current one.
    pub application_version: Option<&'a str>,
    /// The queue it is re-enqueued on. `None` means [`INTERNAL_QUEUE`](crate::sysdb::INTERNAL_QUEUE).
    pub queue_name: Option<&'a str>,
    /// The partition of that queue. `None` clears any key the workflow had.
    pub queue_partition_key: Option<&'a str>,
}

impl RewindOptions<'_> {
    /// Refuses an empty string in any field, as [`ForkOptions::validate`] does and for the same
    /// reason: `Some("")` as a version would replace the workflow's own with an empty one, rather
    /// than leave it alone.
    pub(crate) fn validate(&self) -> Result<(), Error> {
        for (field, value) in [
            ("application_version", &self.application_version),
            ("queue_name", &self.queue_name),
            ("queue_partition_key", &self.queue_partition_key),
        ] {
            if *value == Some("") {
                return Err(Error::InvalidInput {
                    field: field.into(),
                    detail: "must be absent rather than empty".to_owned(),
                });
            }
        }
        Ok(())
    }
}

/// The `DBOS.*` step names the engine records for its own operations.
///
/// **Here rather than in a backend, because they are stored contract.** Each of these lands in
/// `operation_outputs.function_name`, where a replay compares it and a step listing shows it, so
/// they belong beside the row shapes rather than beside the SQL of whichever backend happens to
/// write them. A second backend that redeclared them could drift from this one silently, and
/// nothing would notice until a workflow recorded by one was replayed by the other.
///
/// **`DBOS.`, then camelCase.** The prefix marks a step as the library's own rather than the
/// application's, and the DBOS console relies on it: a step whose name starts with `DBOS.` is shown
/// as a framework action, not as user code. After the prefix, each name is the operation's words
/// in camelCase — `send_bulk` records `DBOS.sendBulk`, and [`DBOS::cancel`](crate::DBOS::cancel)
/// records `DBOS.cancelWorkflow` — the casing the other SDKs sharing the database use for their
/// own steps, and the names Rust 0.5 recorded. Several names are Rust's own or are spelled
/// differently by one SDK, so the casing is shared but not every name is. The console reads some
/// of them by exact name: it tells a parent's await of a child from the child's start by
/// `DBOS.getResult`.
///
/// **Renaming one breaks replay.** A replay compares each recorded name exactly, so a workflow
/// recorded under the old name fails under the new one with [`Error::UnexpectedStep`] and is
/// eventually parked: its step names must read the same in every release that may recover it.
///
/// **Public because a caller may legitimately need to name one** — filtering a step listing to the
/// engine's own operations, or grouping by them — and retyping a string the engine already owns is
/// how the two come apart.
///
/// **Not for asserting the contract, though.** A test that checks a recorded name against the
/// constant the writer used cannot notice the constant changing, which is exactly the change worth
/// noticing: a workflow recorded under the old name fails its replay under the new one with
/// [`Error::UnexpectedStep`]. Those assertions keep their literals on purpose.
pub mod step_names {
    /// The step name `record_sleep` records.
    pub const SLEEP: &str = "DBOS.sleep";

    /// What a patch marker's step name starts with: the marker for the patch named `v2` is
    /// recorded as `DBOS.patch-v2`.
    ///
    /// A marker records that an execution took a [`patch`](crate::patch())'s new path. It is a
    /// row with this name and nothing else — no output, no error, no serialization — at the
    /// position the patch was reached at, which is what a replay looks for there.
    pub const PATCH_PREFIX: &str = "DBOS.patch-";

    /// The step names the two send methods record, one each: `send_message` writes [`SEND`] and
    /// `send_messages` writes [`SEND_BULK`]. The name is therefore the API surface the caller
    /// reached for, never the batch's length — a batch of one is still a batch.
    ///
    /// A method apiece is why: the name comes from which one was called, so nothing has to infer
    /// it. Inferring it from the batch size would record a one-message bulk send as `DBOS.send`,
    /// which is not the call the caller made.
    pub const SEND: &str = "DBOS.send";
    /// What `send_messages` records.
    pub const SEND_BULK: &str = "DBOS.sendBulk";

    /// The step names a stream write records, chosen by whether the value is the closing sentinel.
    ///
    /// Deriving rather than passing means a close is always recorded as a close, whichever entry
    /// point reached it.
    pub const WRITE_STREAM: &str = "DBOS.writeStream";
    /// What a write of the closing sentinel records.
    pub const CLOSE_STREAM: &str = "DBOS.closeStream";

    /// The step names a workflow's stream read records, one step per value it delivers: the
    /// reader's [`READ_STREAM`], and [`READ_STREAM_VALUE`] for the read of a single value.
    ///
    /// The two reads record the same rows under different names so that a workflow changed from
    /// one to the other is caught on replay as `UnexpectedStep` rather than handed a value the
    /// other read recorded.
    pub const READ_STREAM: &str = "DBOS.readStream";
    /// What the read of a single value records.
    pub const READ_STREAM_VALUE: &str = "DBOS.readStreamValue";

    /// The step name `set_event` records, which a replay compares against.
    pub const SET_EVENT: &str = "DBOS.setEvent";

    /// The step name `get_event` records.
    pub const GET_EVENT: &str = "DBOS.getEvent";

    /// The step name a parent writes when it awaits a child.
    ///
    /// Written by
    /// [`record_child_result`](crate::sysdb::SystemDatabase::record_child_result) and read back by
    /// [`check_child_result`](crate::sysdb::SystemDatabase::check_child_result), which is the only
    /// reason both of those exist rather than the caller passing a name.
    pub const GET_RESULT: &str = "DBOS.getResult";

    /// The step name a wait for the *first* of several workflows records, named for the call that
    /// writes it, [`select_workflow`](crate::select_workflow()).
    ///
    /// **There is no constant for the all-wait, because it records nothing.**
    /// [`join_workflows`](crate::join_workflows()) is a plain wait on every surface: it makes no
    /// choice a replay could make differently, and the [`GET_RESULT`] steps it is written to
    /// precede already record the outcomes a replay reads. The free
    /// [`join_workflows`](crate::join_workflows()) sets out the argument.
    pub const SELECT_WORKFLOW: &str = "DBOS.selectWorkflow";

    /// The step name a durable race over steps records, named for the call that writes it,
    /// [`select_step!`](crate::select_step).
    ///
    /// **What it records is a position, not an outcome**: the index of the branch that won, among
    /// the branches that race at this point in the code. The winner's own result is under the
    /// winning step's own id, so recording it here too would keep one outcome in two places.
    pub const SELECT_STEP: &str = "DBOS.selectStep";

    /// The step name `recv` records.
    pub const RECV: &str = "DBOS.recv";

    /// The step a debounce records when a workflow does the bouncing.
    pub const DEBOUNCE: &str = "DBOS.debounceDelayedWorkflow";

    /// The step names the management surface records, which a replay compares against.
    ///
    /// Every one of these is written by a management call made *from inside a workflow* —
    /// [`DBOS::cancel`](crate::DBOS::cancel) and its neighbours — and is what another execution
    /// of that workflow looks the recorded answer up by.
    ///
    /// **The singular name covers the bulk form too.** The singular forms *are* the bulk ones with
    /// a single id, so a workflow that switches between [`cancel`](crate::DBOS::cancel) and
    /// [`cancel_all`](crate::DBOS::cancel_all) between runs still replays instead of raising
    /// [`Error::UnexpectedStep`](crate::sysdb::Error::UnexpectedStep).
    ///
    /// **A queues-only listing records [`LIST_WORKFLOWS`] like any other**, because
    /// [`WorkflowFilter::queues_only`](super::WorkflowFilter::queues_only) is a filter on the one
    /// listing call rather than a second entry point.
    ///
    /// **Nothing names a retrieve.** [`retrieve_workflow`](crate::DBOS::retrieve_workflow) does no
    /// I/O, and a call that reads nothing has nothing to replay.
    ///
    /// All of these are recorded from down here rather than by the engine, because each
    /// checkpoint commits in the same transaction as the operation it records — see
    /// [`fork_workflows`](crate::sysdb::SystemDatabase::fork_workflows).
    ///
    /// **[`FORK_WORKFLOW`] covers every fork point**, because
    /// [`fork_from`](crate::sysdb::SystemDatabase::fork_from) resolves all four
    /// [`ForkPoint`](super::ForkPoint)s through one method.
    pub const CANCEL_WORKFLOW: &str = "DBOS.cancelWorkflow";
    /// What resuming a workflow records.
    pub const RESUME_WORKFLOW: &str = "DBOS.resumeWorkflow";
    /// What deleting a workflow records.
    pub const DELETE_WORKFLOW: &str = "DBOS.deleteWorkflow";
    /// What forking a workflow records.
    pub const FORK_WORKFLOW: &str = "DBOS.forkWorkflow";
    /// What rewinding a workflow records.
    pub const REWIND_WORKFLOW: &str = "DBOS.rewindWorkflow";
    /// What changing a workflow's delay records.
    pub const SET_WORKFLOW_DELAY: &str = "DBOS.setWorkflowDelay";
    /// What updating a workflow's attributes records.
    pub const UPDATE_WORKFLOW_ATTRIBUTES: &str = "DBOS.updateWorkflowAttributes";
    /// What listing workflows records.
    pub const LIST_WORKFLOWS: &str = "DBOS.listWorkflows";
    /// What listing a workflow's steps records.
    pub const LIST_WORKFLOW_STEPS: &str = "DBOS.listWorkflowSteps";

    /// The step names the schedule methods record, which a replay compares against.
    ///
    /// Pausing and resuming reach one method but record two names: the name follows the status
    /// being set rather than the method being called.
    pub const CREATE_SCHEDULE: &str = "DBOS.createSchedule";
    /// What creating or replacing a schedule records.
    pub const UPSERT_SCHEDULE: &str = "DBOS.upsertSchedule";
    /// What reading a schedule records.
    pub const GET_SCHEDULE: &str = "DBOS.getSchedule";
    /// What listing schedules records.
    pub const LIST_SCHEDULES: &str = "DBOS.listSchedules";
    /// What updating a schedule records.
    pub const UPDATE_SCHEDULE: &str = "DBOS.updateSchedule";
    /// What pausing a schedule records.
    pub const PAUSE_SCHEDULE: &str = "DBOS.pauseSchedule";
    /// What resuming a schedule records.
    pub const RESUME_SCHEDULE: &str = "DBOS.resumeSchedule";
    /// What deleting a schedule records.
    pub const DELETE_SCHEDULE: &str = "DBOS.deleteSchedule";
}

/// One message to deliver to a workflow.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Message<'a> {
    /// The workflow it is for.
    pub destination_id: &'a str,
    /// The topic it is filed under, or `None` for the untopicked default.
    ///
    /// `None` is stored as a sentinel string rather than `NULL`, because a receiver selects on
    /// equality and `topic = NULL` matches nothing. Other SDKs sharing the database use the same
    /// sentinel, so it is part of the cross-SDK contract rather than an encoding choice.
    pub topic: Option<&'a str>,
    /// The payload, already encoded by the caller.
    pub message: &'a str,
    /// A key that makes re-sending this message a no-op.
    ///
    /// Absent, each send is a distinct message. Present, it becomes the row's primary key, so a
    /// second send with the same key is discarded by the database rather than delivered twice.
    pub idempotency_key: Option<&'a str>,
}

/// Who wrote to a stream, which decides whether the write is itself a durable step.
///
/// A workflow's write and a step's write differ in nothing else. Naming the difference instead of
/// duplicating the method is the same choice made for
/// [`send_messages`](crate::sysdb::SystemDatabase::send_messages): the split is in the behaviour,
/// not the operation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WrittenBy {
    /// The workflow body itself. The write *is* a step: it is recorded, and a replay finds it
    /// and writes nothing rather than appending a second entry.
    Workflow,
    /// Code running inside a step. The enclosing step is already the durable unit, so this
    /// records nothing of its own — a step that reruns rewrites its stream entries, which is
    /// what makes the step the thing being replayed.
    Step,
}

/// Whether a schedule fires.
///
/// Two states. Pausing does not delete the row or forget [`ScheduleRecord::last_fired_at`], so
/// resuming picks up where it left off.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "UPPERCASE")]
pub enum ScheduleStatus {
    /// Firing on its cron expression.
    Active,
    /// Registered but not firing.
    Paused,
}

impl ScheduleStatus {
    /// The stored spelling.
    pub fn as_str(self) -> &'static str {
        match self {
            ScheduleStatus::Active => "ACTIVE",
            ScheduleStatus::Paused => "PAUSED",
        }
    }

    /// Parses a stored value, for the reason [`WorkflowStatus::parse`] returns an option.
    pub fn parse(value: &str) -> Option<Self> {
        Some(match value {
            "ACTIVE" => ScheduleStatus::Active,
            "PAUSED" => ScheduleStatus::Paused,
            _ => return None,
        })
    }
}

/// A registered schedule, as stored.
///
/// The row carries both a *definition* — the cron expression, the workflow it fires, its context
/// and queue — and *runtime state*: the status and when it last fired. Which half a write may
/// touch is the distinction the update methods are built around, so a redeployment that re-applies
/// an unchanged definition does not restart a schedule or forget where it had got to.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct ScheduleRecord {
    /// Generated identity, distinct from the name. Preserved across re-registration.
    pub schedule_id: String,
    /// The schedule's name, which is its address — unique across the whole table.
    pub schedule_name: String,
    /// The registered function the schedule fires.
    pub workflow_name: String,
    /// The class that function belongs to, for class-bound workflows.
    pub workflow_class_name: Option<String>,
    /// The cron expression, uninterpreted. This layer stores it and never parses it.
    pub schedule: String,
    /// Whether the schedule fires.
    pub status: ScheduleStatus,
    /// The encoded context handed to each firing, opaque here as every payload is.
    pub context: String,
    /// When the schedule last fired.
    ///
    /// Stored as ISO-8601 text rather than the epoch milliseconds every other time column holds,
    /// and read back through [`Timestamp::parse_iso8601`] so a caller gets an instant either way.
    /// Writers sharing the database spell it differently — see
    /// [`SystemDatabase::update_schedule_last_fired_at`](crate::sysdb::SystemDatabase::update_schedule_last_fired_at).
    pub last_fired_at: Option<Timestamp>,
    /// Whether missed firings are made up when a paused or stopped schedule resumes.
    pub automatic_backfill: bool,
    /// The timezone the cron expression is read in, or `None` for UTC.
    pub cron_timezone: Option<String>,
    /// The queue firings are enqueued onto, or `None` for [`INTERNAL_QUEUE`](crate::sysdb::INTERNAL_QUEUE).
    pub queue_name: Option<String>,
    /// The application that owns the schedule, or `None` if it is unclaimed.
    pub application_name: Option<String>,
}

/// A schedule to register, as the caller supplies it.
///
/// Borrowed and separate from [`ScheduleRecord`] for the reason [`NewQueue`] is separate from
/// [`QueueRecord`]: registering does not require an owner, and the record always reports one.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NewSchedule<'a> {
    /// See [`ScheduleRecord::schedule_id`]. `None` generates a UUID; a supplied id is stored as
    /// given, so it need not be one.
    ///
    /// The fallback lives here, as it does for `application_name`, so this layer stands on its
    /// own, and a registration layer above it that supplies the id makes this a second line of
    /// defence rather than the only one.
    pub schedule_id: Option<&'a str>,
    /// See [`ScheduleRecord::schedule_name`].
    pub schedule_name: &'a str,
    /// See [`ScheduleRecord::workflow_name`].
    pub workflow_name: &'a str,
    /// See [`ScheduleRecord::workflow_class_name`].
    pub workflow_class_name: Option<&'a str>,
    /// See [`ScheduleRecord::schedule`].
    pub schedule: &'a str,
    /// See [`ScheduleRecord::status`].
    pub status: ScheduleStatus,
    /// See [`ScheduleRecord::context`].
    pub context: &'a str,
    /// See [`ScheduleRecord::last_fired_at`]. Only used on a fresh insert: the upsert's conflict
    /// clause keeps the stored value, so re-registering a schedule cannot rewind it.
    pub last_fired_at: Option<Timestamp>,
    /// See [`ScheduleRecord::automatic_backfill`].
    pub automatic_backfill: bool,
    /// See [`ScheduleRecord::cron_timezone`].
    pub cron_timezone: Option<&'a str>,
    /// See [`ScheduleRecord::queue_name`].
    pub queue_name: Option<&'a str>,
    /// The application to register the schedule for; `None` means the writing handle's own.
    pub application_name: Option<&'a str>,
}

impl<'a> NewSchedule<'a> {
    /// An active schedule with no context, firing the named workflow on the given expression.
    pub fn new(schedule_name: &'a str, workflow_name: &'a str, schedule: &'a str) -> Self {
        Self {
            schedule_id: None,
            schedule_name,
            workflow_name,
            workflow_class_name: None,
            schedule,
            status: ScheduleStatus::Active,
            context: "null",
            last_fired_at: None,
            automatic_backfill: false,
            cron_timezone: None,
            queue_name: None,
            application_name: None,
        }
    }
}

/// The fields of a registered schedule that a partial update may change.
///
/// **Definition only.** The identity, the status and the last firing are runtime state, moved by
/// their own methods, so re-applying a definition cannot silently restart a schedule or forget
/// where it had got to. The type makes that unrepresentable rather than merely documented.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ScheduleUpdate<'a> {
    /// See [`ScheduleRecord::schedule`].
    pub schedule: Change<&'a str>,
    /// See [`ScheduleRecord::context`].
    pub context: Change<&'a str>,
    /// See [`ScheduleRecord::automatic_backfill`].
    pub automatic_backfill: Change<bool>,
    /// See [`ScheduleRecord::cron_timezone`].
    pub cron_timezone: Change<Option<&'a str>>,
    /// See [`ScheduleRecord::queue_name`].
    pub queue_name: Change<Option<&'a str>>,
}

impl ScheduleUpdate<'_> {
    /// Whether this would change nothing.
    pub fn is_empty(&self) -> bool {
        self.schedule.is_leave()
            && self.context.is_leave()
            && self.automatic_backfill.is_leave()
            && self.cron_timezone.is_leave()
            && self.queue_name.is_leave()
    }
}

/// Which schedules a listing returns.
///
/// Every field narrows; an empty filter returns the table. `app_scope` takes the
/// *observability* form — unset means the reading handle's own plus the unclaimed — because this
/// is a search rather than an addressed read.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ScheduleFilter<'a> {
    /// Only these statuses, or every status when empty.
    pub statuses: Vec<ScheduleStatus>,
    /// Only schedules firing these workflows, or every workflow when empty.
    pub workflow_names: Vec<&'a str>,
    /// Only schedules whose name starts with one of these, or every name when empty.
    pub schedule_name_prefixes: Vec<&'a str>,
    /// Which applications' schedules to return.
    pub app_scope: AppScope<'a>,
}