oxide_update_engine_types/

events.rs

// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at https://mozilla.org/MPL/2.0/.

//! Progress, success and failure events generated by the update engine.
//!
//! Events are serializable. When the `schemars08` feature is enabled,
//! they also implement `JsonSchema` so that they can be transmitted
//! over the wire with schema support.

#[cfg(feature = "schemars08")]
use crate::spec::JsonSchemaEngineSpec;
use crate::{
    errors::ConvertGenericError,
    spec::{EngineSpec, GenericSpec},
};
use derive_where::derive_where;
use newtype_uuid::{TypedUuid, TypedUuidKind, TypedUuidTag};
use serde::{Deserialize, Deserializer, Serialize, de::IgnoredAny};
use std::{borrow::Cow, fmt, time::Duration};

/// Marker type for [`ExecutionUuid`].
pub enum ExecutionUuidKind {}

impl TypedUuidKind for ExecutionUuidKind {
    #[inline]
    fn tag() -> TypedUuidTag {
        const TAG: TypedUuidTag = TypedUuidTag::new("execution_id");
        TAG
    }

    #[inline]
    fn alias() -> Option<&'static str> {
        // Used by the schemars integration to produce schema name
        // "ExecutionUuid" (matching the type alias).
        Some("ExecutionUuid")
    }
}

#[cfg(feature = "schemars08")]
impl schemars::JsonSchema for ExecutionUuidKind {
    fn schema_name() -> String {
        "ExecutionUuidKind".to_owned()
    }

    fn json_schema(
        _: &mut schemars::r#gen::SchemaGenerator,
    ) -> schemars::schema::Schema {
        use crate::schema::{EVENTS_MODULE, rust_type_for_events};

        // Produce a UUID schema with x-rust-type so that
        // newtype-uuid's lifting mechanism produces a
        // TypedUuid<ExecutionUuidKind> schema with path
        // "oxide_update_engine_types::events::ExecutionUuid" (via
        // the alias() method on TypedUuidKind).
        schemars::schema::SchemaObject {
            instance_type: Some(schemars::schema::InstanceType::String.into()),
            format: Some("uuid".to_owned()),
            extensions: [(
                "x-rust-type".to_owned(),
                rust_type_for_events(&format!(
                    "{EVENTS_MODULE}::ExecutionUuidKind"
                )),
            )]
            .into_iter()
            .collect(),
            ..Default::default()
        }
        .into()
    }
}

/// A unique identifier for an execution of an update engine.
///
/// Each time an `UpdateEngine` is executed, it is assigned a unique
/// `ExecutionUuid`.
pub type ExecutionUuid = TypedUuid<ExecutionUuidKind>;

#[derive_where(Clone, Debug, PartialEq, Eq)]
pub enum Event<S: EngineSpec> {
    Step(StepEvent<S>),
    Progress(ProgressEvent<S>),
}

impl<S: EngineSpec> Event<S> {
    /// Converts a generic version into self.
    ///
    /// This version can be used to convert a generic type into a more concrete
    /// form.
    pub fn from_generic(
        value: Event<GenericSpec>,
    ) -> Result<Self, ConvertGenericError> {
        let ret = match value {
            Event::Step(event) => Event::Step(StepEvent::from_generic(event)?),
            Event::Progress(event) => {
                Event::Progress(ProgressEvent::from_generic(event)?)
            }
        };
        Ok(ret)
    }

    /// Converts self into its generic version.
    ///
    /// This version can be used to share data across different kinds of
    /// engines.
    ///
    /// If any of the data in self fails to serialize to a
    /// [`serde_json::Value`], it will be replaced with
    /// [`serde_json::Value::Null`]. Since `serde_json::Value` represents
    /// an arbitrary JSON value, such data would have failed to serialize
    /// anyway.
    pub fn into_generic(self) -> Event<GenericSpec> {
        match self {
            Event::Step(event) => Event::Step(event.into_generic()),
            Event::Progress(event) => Event::Progress(event.into_generic()),
        }
    }
}

#[derive(Deserialize, Serialize)]
#[derive_where(Clone, Debug, Eq, PartialEq)]
#[serde(bound = "", rename_all = "snake_case")]
pub struct StepEvent<S: EngineSpec> {
    /// The specification that this event belongs to.
    ///
    /// This is typically the name of the type `S` for which `EngineSpec` is
    /// implemented.
    ///
    /// This can be used along with `Self::from_generic` to identify which
    /// specification to deserialize generic metadata against. For example:
    ///
    /// ```rust,ignore
    /// if event.spec == "MySpec" {
    ///     // event is likely generated from a MySpec engine.
    ///     let event = Event::<MySpec>::from_generic(event)?;
    ///     // ...
    /// }
    /// ```
    pub spec: String,

    /// The execution ID.
    pub execution_id: ExecutionUuid,

    /// A monotonically increasing index for this `StepEvent`.
    pub event_index: usize,

    /// Total time elapsed since the start of execution.
    pub total_elapsed: Duration,

    /// The kind of event this is.
    #[serde(rename = "data")]
    pub kind: StepEventKind<S>,
}

#[cfg(feature = "schemars08")]
impl<S: JsonSchemaEngineSpec> schemars::JsonSchema for StepEvent<S> {
    fn schema_name() -> String {
        format!("StepEventFor{}", S::schema_name())
    }

    fn json_schema(
        generator: &mut schemars::r#gen::SchemaGenerator,
    ) -> schemars::schema::Schema {
        use crate::schema::with_description;
        use schemars::schema::{ObjectValidation, SchemaObject};

        let mut obj = ObjectValidation::default();
        obj.properties.insert(
            "spec".to_owned(),
            with_description(
                generator.subschema_for::<String>(),
                "The specification that this event belongs to.",
            ),
        );
        obj.properties.insert(
            "execution_id".to_owned(),
            with_description(
                generator.subschema_for::<ExecutionUuid>(),
                "The execution ID.",
            ),
        );
        obj.properties.insert(
            "event_index".to_owned(),
            with_description(
                generator.subschema_for::<usize>(),
                "A monotonically increasing index for this \
                 `StepEvent`.",
            ),
        );
        obj.properties.insert(
            "total_elapsed".to_owned(),
            with_description(
                generator.subschema_for::<Duration>(),
                "Total time elapsed since the start of execution.",
            ),
        );
        obj.properties.insert(
            "data".to_owned(),
            with_description(
                generator.subschema_for::<StepEventKind<S>>(),
                "The kind of event this is.",
            ),
        );
        obj.required =
            ["spec", "execution_id", "event_index", "total_elapsed", "data"]
                .into_iter()
                .map(String::from)
                .collect();

        let mut extensions = serde_json::Map::new();
        if let Some(info) = S::rust_type_info() {
            extensions.insert(
                "x-rust-type".to_owned(),
                crate::schema::rust_type_for_generic(&info, "StepEvent"),
            );
        }

        SchemaObject {
            instance_type: Some(schemars::schema::InstanceType::Object.into()),
            object: Some(Box::new(obj)),
            extensions: extensions.into_iter().collect(),
            ..Default::default()
        }
        .into()
    }
}

impl<S: EngineSpec> StepEvent<S> {
    /// Returns a progress event associated with this step event, if any.
    ///
    /// Some step events have an implicit progress event of kind
    /// [`ProgressEventKind::WaitingForProgress`] associated with them. This
    /// causes those step events to generate progress events.
    pub fn progress_event(&self) -> Option<ProgressEvent<S>> {
        match &self.kind {
            StepEventKind::ExecutionStarted { first_step, .. } => {
                Some(ProgressEvent {
                    spec: self.spec.clone(),
                    execution_id: self.execution_id,
                    total_elapsed: self.total_elapsed,
                    kind: ProgressEventKind::WaitingForProgress {
                        step: first_step.clone(),
                        attempt: 1,
                        step_elapsed: Duration::ZERO,
                        attempt_elapsed: Duration::ZERO,
                    },
                })
            }
            StepEventKind::ProgressReset {
                step,
                attempt,
                step_elapsed,
                attempt_elapsed,
                ..
            } => Some(ProgressEvent {
                spec: self.spec.clone(),
                execution_id: self.execution_id,
                total_elapsed: self.total_elapsed,
                kind: ProgressEventKind::WaitingForProgress {
                    step: step.clone(),
                    attempt: *attempt,
                    step_elapsed: *step_elapsed,
                    attempt_elapsed: *attempt_elapsed,
                },
            }),
            StepEventKind::AttemptRetry {
                step,
                next_attempt,
                step_elapsed,
                ..
            } => Some(ProgressEvent {
                spec: self.spec.clone(),
                execution_id: self.execution_id,
                total_elapsed: self.total_elapsed,
                kind: ProgressEventKind::WaitingForProgress {
                    step: step.clone(),
                    attempt: *next_attempt,
                    step_elapsed: *step_elapsed,
                    // For this attempt, zero time has passed so far.
                    attempt_elapsed: Duration::ZERO,
                },
            }),
            StepEventKind::StepCompleted { next_step, .. } => {
                Some(ProgressEvent {
                    spec: self.spec.clone(),
                    execution_id: self.execution_id,
                    total_elapsed: self.total_elapsed,
                    kind: ProgressEventKind::WaitingForProgress {
                        step: next_step.clone(),
                        attempt: 1,
                        // For this next step, zero time has passed so far.
                        step_elapsed: Duration::ZERO,
                        attempt_elapsed: Duration::ZERO,
                    },
                })
            }
            StepEventKind::Nested {
                step,
                attempt,
                step_elapsed,
                attempt_elapsed,
                event,
                ..
            } => event.progress_event().map(|progress_event| ProgressEvent {
                spec: self.spec.clone(),
                execution_id: self.execution_id,
                total_elapsed: self.total_elapsed,
                kind: ProgressEventKind::Nested {
                    step: step.clone(),
                    attempt: *attempt,
                    event: Box::new(progress_event),
                    step_elapsed: *step_elapsed,
                    attempt_elapsed: *attempt_elapsed,
                },
            }),
            StepEventKind::NoStepsDefined
            | StepEventKind::ExecutionCompleted { .. }
            | StepEventKind::ExecutionFailed { .. }
            | StepEventKind::ExecutionAborted { .. }
            | StepEventKind::Unknown => None,
        }
    }

    /// Returns the execution ID for the leaf event, recursing into nested
    /// events if necessary.
    pub fn leaf_execution_id(&self) -> ExecutionUuid {
        match &self.kind {
            StepEventKind::Nested { event, .. } => event.leaf_execution_id(),
            _ => self.execution_id,
        }
    }

    /// Returns the event index for the leaf event, recursing into nested events
    /// if necessary.
    pub fn leaf_event_index(&self) -> usize {
        match &self.kind {
            StepEventKind::Nested { event, .. } => event.leaf_event_index(),
            _ => self.event_index,
        }
    }

    /// Converts a generic version into self.
    ///
    /// This version can be used to convert a generic type into a more concrete
    /// form.
    pub fn from_generic(
        value: StepEvent<GenericSpec>,
    ) -> Result<Self, ConvertGenericError> {
        Ok(StepEvent {
            spec: value.spec,
            execution_id: value.execution_id,
            event_index: value.event_index,
            total_elapsed: value.total_elapsed,
            kind: StepEventKind::from_generic(value.kind)
                .map_err(|error| error.parent("kind"))?,
        })
    }

    /// Converts self into its generic version.
    ///
    /// This version can be used to share data across different kinds of
    /// engines.
    ///
    /// If any of the data in self fails to serialize to a
    /// [`serde_json::Value`], it will be replaced with
    /// [`serde_json::Value::Null`]. Since `serde_json::Value` represents
    /// an arbitrary JSON value, such data would have failed to serialize
    /// anyway.
    pub fn into_generic(self) -> StepEvent<GenericSpec> {
        StepEvent {
            spec: self.spec,
            execution_id: self.execution_id,
            event_index: self.event_index,
            total_elapsed: self.total_elapsed,
            kind: self.kind.into_generic(),
        }
    }
}

#[derive(Deserialize, Serialize)]
#[cfg_attr(feature = "schemars08", derive(schemars::JsonSchema))]
#[derive_where(Clone, Debug, Eq, PartialEq)]
#[serde(bound = "", rename_all = "snake_case", tag = "kind")]
#[cfg_attr(
    feature = "schemars08",
    schemars(
        rename = "StepEventKindFor{S}",
        bound = "S: JsonSchemaEngineSpec",
    )
)]
pub enum StepEventKind<S: EngineSpec> {
    /// No steps were defined, and the executor exited without doing anything.
    ///
    /// This is a terminal event: it is guaranteed that no more events will be
    /// seen after this one.
    NoStepsDefined,

    /// Execution was started.
    ///
    /// This is an initial event -- it is always expected to be the first event
    /// received from the event stream.
    ExecutionStarted {
        /// The list of steps that will be executed.
        steps: Vec<StepInfo<S>>,

        /// A list of components, along with the number of items each component has.
        components: Vec<StepComponentSummary<S>>,

        /// Information about the first step.
        first_step: StepInfoWithMetadata<S>,
    },

    /// Progress was reset along an attempt, and this attempt is going down a
    /// different path.
    ProgressReset {
        /// Information about the step.
        step: StepInfoWithMetadata<S>,

        /// The current attempt number.
        attempt: usize,

        /// Progress-related metadata associated with this attempt.
        metadata: S::ProgressMetadata,

        /// Total time elapsed since the start of the step. Includes prior
        /// attempts.
        step_elapsed: Duration,

        /// The amount of time this attempt has taken so far.
        attempt_elapsed: Duration,

        /// A message associated with the reset.
        message: Cow<'static, str>,
    },

    /// An attempt failed and this step is being retried.
    AttemptRetry {
        /// Information about the step.
        step: StepInfoWithMetadata<S>,

        /// The attempt number for the next attempt.
        next_attempt: usize,

        /// Total time elapsed since the start of the step. Includes prior
        /// attempts.
        step_elapsed: Duration,

        /// The amount of time the previous attempt took.
        attempt_elapsed: Duration,

        /// A message associated with the retry.
        message: Cow<'static, str>,
    },

    /// A step is complete and the next step has been started.
    StepCompleted {
        /// Information about the step that just completed.
        step: StepInfoWithMetadata<S>,

        /// The attempt number that completed.
        attempt: usize,

        /// The outcome of the step.
        outcome: StepOutcome<S>,

        /// The next step that is being started.
        next_step: StepInfoWithMetadata<S>,

        /// Total time elapsed since the start of the step. Includes prior
        /// attempts.
        step_elapsed: Duration,

        /// The time it took for this attempt to complete.
        attempt_elapsed: Duration,
    },

    /// Execution is complete.
    ///
    /// This is a terminal event: it is guaranteed that no more events will be
    /// seen after this one.
    ExecutionCompleted {
        /// Information about the last step that completed.
        last_step: StepInfoWithMetadata<S>,

        /// The attempt number that completed.
        last_attempt: usize,

        /// The outcome of the last step.
        last_outcome: StepOutcome<S>,

        /// Total time elapsed since the start of the step. Includes prior
        /// attempts.
        step_elapsed: Duration,

        /// The time it took for this attempt to complete.
        attempt_elapsed: Duration,
    },

    /// Execution failed.
    ///
    /// This is a terminal event: it is guaranteed that no more events will be
    /// seen after this one.
    ExecutionFailed {
        /// Information about the step that failed.
        failed_step: StepInfoWithMetadata<S>,

        /// The total number of attempts that were performed before the step failed.
        total_attempts: usize,

        /// Total time elapsed since the start of the step. Includes prior
        /// attempts.
        step_elapsed: Duration,

        /// The time it took for this attempt to complete.
        attempt_elapsed: Duration,

        /// A message associated with the failure.
        message: String,

        /// A chain of causes associated with the failure.
        causes: Vec<String>,
    },

    /// Execution aborted by an external user.
    ///
    /// This is a terminal event: it is guaranteed that no more events will be
    /// seen after this one.
    ExecutionAborted {
        /// Information about the step that was running at the time execution
        /// was aborted.
        aborted_step: StepInfoWithMetadata<S>,

        /// The attempt that was running at the time the step was aborted.
        attempt: usize,

        /// Total time elapsed since the start of the step. Includes prior
        /// attempts.
        step_elapsed: Duration,

        /// The time it took for this attempt to complete.
        attempt_elapsed: Duration,

        /// A message associated with the abort.
        message: String,
    },

    /// A nested step event occurred.
    Nested {
        /// Information about the step that's occurring.
        step: StepInfoWithMetadata<S>,

        /// The current attempt number.
        attempt: usize,

        /// The event that occurred.
        event: Box<StepEvent<GenericSpec>>,

        /// Total time elapsed since the start of the step. Includes prior
        /// attempts.
        step_elapsed: Duration,

        /// The time it took for this attempt to complete.
        attempt_elapsed: Duration,
    },

    /// Future variants that might be unknown.
    #[serde(other, deserialize_with = "deserialize_ignore_any")]
    Unknown,
}

fn deserialize_ignore_any<'de, D: Deserializer<'de>, T: Default>(
    deserializer: D,
) -> Result<T, D::Error> {
    IgnoredAny::deserialize(deserializer).map(|_| T::default())
}

impl<S: EngineSpec> StepEventKind<S> {
    /// Returns whether this is a terminal step event.
    ///
    /// Terminal events guarantee that there are no further events coming from
    /// this update engine.
    ///
    /// This does not recurse into nested events; those are always non-terminal.
    pub fn is_terminal(&self) -> StepEventIsTerminal {
        match self {
            StepEventKind::NoStepsDefined
            | StepEventKind::ExecutionCompleted { .. } => {
                StepEventIsTerminal::Terminal { success: true }
            }
            StepEventKind::ExecutionFailed { .. }
            | StepEventKind::ExecutionAborted { .. } => {
                StepEventIsTerminal::Terminal { success: false }
            }
            StepEventKind::ExecutionStarted { .. }
            | StepEventKind::ProgressReset { .. }
            | StepEventKind::AttemptRetry { .. }
            | StepEventKind::StepCompleted { .. }
            | StepEventKind::Nested { .. }
            | StepEventKind::Unknown => StepEventIsTerminal::NonTerminal,
        }
    }

    /// Returns the priority of the event.
    ///
    /// For more about this, see [`StepEventPriority`].
    pub fn priority(&self) -> StepEventPriority {
        match self {
            StepEventKind::NoStepsDefined
            | StepEventKind::ExecutionStarted { .. }
            | StepEventKind::StepCompleted { .. }
            | StepEventKind::ExecutionCompleted { .. }
            | StepEventKind::ExecutionFailed { .. }
            | StepEventKind::ExecutionAborted { .. } => StepEventPriority::High,
            StepEventKind::ProgressReset { .. }
            | StepEventKind::AttemptRetry { .. }
            | StepEventKind::Unknown => StepEventPriority::Low,
            StepEventKind::Nested { event, .. } => event.kind.priority(),
        }
    }

    /// Converts a generic version into self.
    ///
    /// This version can be used to convert a generic type into a more concrete
    /// form.
    pub fn from_generic(
        value: StepEventKind<GenericSpec>,
    ) -> Result<Self, ConvertGenericError> {
        let ret = match value {
            StepEventKind::NoStepsDefined => StepEventKind::NoStepsDefined,
            StepEventKind::ExecutionStarted {
                steps,
                components,
                first_step,
            } => StepEventKind::ExecutionStarted {
                steps: steps
                    .into_iter()
                    .enumerate()
                    .map(|(index, step)| {
                        StepInfo::from_generic(step)
                            .map_err(|error| error.parent_array("steps", index))
                    })
                    .collect::<Result<Vec<_>, _>>()?,
                components: components
                    .into_iter()
                    .enumerate()
                    .map(|(index, component)| {
                        StepComponentSummary::from_generic(component).map_err(
                            |error| error.parent_array("components", index),
                        )
                    })
                    .collect::<Result<Vec<_>, _>>()?,
                first_step: StepInfoWithMetadata::from_generic(first_step)
                    .map_err(|error| error.parent("first_step"))?,
            },
            StepEventKind::ProgressReset {
                step,
                attempt,
                metadata,
                step_elapsed,
                attempt_elapsed,
                message,
            } => StepEventKind::ProgressReset {
                step: StepInfoWithMetadata::from_generic(step)
                    .map_err(|error| error.parent("step"))?,
                attempt,
                metadata: serde_json::from_value(metadata).map_err(
                    |error| ConvertGenericError::new("metadata", error),
                )?,
                step_elapsed,
                attempt_elapsed,
                message,
            },
            StepEventKind::AttemptRetry {
                step,
                next_attempt,
                step_elapsed,
                attempt_elapsed,
                message,
            } => StepEventKind::AttemptRetry {
                step: StepInfoWithMetadata::from_generic(step)
                    .map_err(|error| error.parent("step"))?,
                next_attempt,
                step_elapsed,
                attempt_elapsed,
                message,
            },
            StepEventKind::StepCompleted {
                step,
                attempt,
                outcome,
                next_step,
                step_elapsed,
                attempt_elapsed,
            } => StepEventKind::StepCompleted {
                step: StepInfoWithMetadata::from_generic(step)
                    .map_err(|error| error.parent("step"))?,
                attempt,
                outcome: StepOutcome::from_generic(outcome)
                    .map_err(|error| error.parent("outcome"))?,
                next_step: StepInfoWithMetadata::from_generic(next_step)
                    .map_err(|error| error.parent("next_step"))?,
                step_elapsed,
                attempt_elapsed,
            },
            StepEventKind::ExecutionCompleted {
                last_step,
                last_attempt,
                last_outcome,
                step_elapsed,
                attempt_elapsed,
            } => StepEventKind::ExecutionCompleted {
                last_step: StepInfoWithMetadata::from_generic(last_step)
                    .map_err(|error| error.parent("last_step"))?,
                last_attempt,
                last_outcome: StepOutcome::from_generic(last_outcome)
                    .map_err(|error| error.parent("last_outcome"))?,
                step_elapsed,
                attempt_elapsed,
            },
            StepEventKind::ExecutionFailed {
                failed_step,
                total_attempts,
                step_elapsed,
                attempt_elapsed,
                message,
                causes,
            } => StepEventKind::ExecutionFailed {
                failed_step: StepInfoWithMetadata::from_generic(failed_step)
                    .map_err(|error| error.parent("failed_step"))?,
                total_attempts,
                step_elapsed,
                attempt_elapsed,
                message,
                causes,
            },
            StepEventKind::ExecutionAborted {
                aborted_step,
                attempt,
                step_elapsed,
                attempt_elapsed,
                message,
            } => StepEventKind::ExecutionAborted {
                aborted_step: StepInfoWithMetadata::from_generic(aborted_step)
                    .map_err(|error| error.parent("aborted_step"))?,
                attempt,
                step_elapsed,
                attempt_elapsed,
                message,
            },
            StepEventKind::Nested {
                step,
                attempt,
                event,
                step_elapsed,
                attempt_elapsed,
            } => StepEventKind::Nested {
                step: StepInfoWithMetadata::from_generic(step)
                    .map_err(|error| error.parent("step"))?,
                attempt,
                event,
                step_elapsed,
                attempt_elapsed,
            },
            StepEventKind::Unknown => StepEventKind::Unknown,
        };
        Ok(ret)
    }

    /// Converts self into its generic version.
    ///
    /// This version can be used to share data across different kinds of
    /// engines.
    ///
    /// If any of the data in self fails to serialize to a
    /// [`serde_json::Value`], it will be replaced with
    /// [`serde_json::Value::Null`]. Since `serde_json::Value` represents
    /// an arbitrary JSON value, such data would have failed to serialize
    /// anyway.
    pub fn into_generic(self) -> StepEventKind<GenericSpec> {
        match self {
            StepEventKind::NoStepsDefined => StepEventKind::NoStepsDefined,
            StepEventKind::ExecutionStarted {
                steps,
                components,
                first_step,
            } => StepEventKind::ExecutionStarted {
                steps: steps
                    .into_iter()
                    .map(|step| step.into_generic())
                    .collect(),
                components: components
                    .into_iter()
                    .map(|component| component.into_generic())
                    .collect(),
                first_step: first_step.into_generic(),
            },
            StepEventKind::ProgressReset {
                step,
                attempt,
                metadata,
                step_elapsed,
                attempt_elapsed,
                message,
            } => StepEventKind::ProgressReset {
                step: step.into_generic(),
                attempt,
                metadata: serde_json::to_value(metadata)
                    .unwrap_or(serde_json::Value::Null),
                step_elapsed,
                attempt_elapsed,
                message,
            },
            StepEventKind::AttemptRetry {
                step,
                next_attempt,
                step_elapsed,
                attempt_elapsed,
                message,
            } => StepEventKind::AttemptRetry {
                step: step.into_generic(),
                next_attempt,
                step_elapsed,
                attempt_elapsed,
                message,
            },
            StepEventKind::StepCompleted {
                step,
                attempt,
                outcome,
                next_step,
                step_elapsed,
                attempt_elapsed,
            } => StepEventKind::StepCompleted {
                step: step.into_generic(),
                attempt,
                outcome: outcome.into_generic(),
                next_step: next_step.into_generic(),
                step_elapsed,
                attempt_elapsed,
            },
            StepEventKind::ExecutionCompleted {
                last_step,
                last_attempt,
                last_outcome,
                step_elapsed,
                attempt_elapsed,
            } => StepEventKind::ExecutionCompleted {
                last_step: last_step.into_generic(),
                last_attempt,
                last_outcome: last_outcome.into_generic(),
                step_elapsed,
                attempt_elapsed,
            },
            StepEventKind::ExecutionFailed {
                failed_step,
                total_attempts,
                step_elapsed,
                attempt_elapsed,
                message,
                causes,
            } => StepEventKind::ExecutionFailed {
                failed_step: failed_step.into_generic(),
                total_attempts,
                step_elapsed,
                attempt_elapsed,
                message,
                causes,
            },
            StepEventKind::ExecutionAborted {
                aborted_step,
                attempt,
                step_elapsed,
                attempt_elapsed,
                message,
            } => StepEventKind::ExecutionAborted {
                aborted_step: aborted_step.into_generic(),
                attempt,
                step_elapsed,
                attempt_elapsed,
                message,
            },
            StepEventKind::Nested {
                step,
                attempt,
                event,
                step_elapsed,
                attempt_elapsed,
            } => StepEventKind::Nested {
                step: step.into_generic(),
                attempt,
                event,
                step_elapsed,
                attempt_elapsed,
            },
            StepEventKind::Unknown => StepEventKind::Unknown,
        }
    }

    /// If this represents a successfully-completed step, returns the outcome.
    ///
    /// This does not recurse into nested events.
    pub fn step_outcome(&self) -> Option<&StepOutcome<S>> {
        match self {
            StepEventKind::StepCompleted { outcome, .. }
            | StepEventKind::ExecutionCompleted {
                last_outcome: outcome, ..
            } => Some(outcome),
            StepEventKind::NoStepsDefined
            | StepEventKind::ExecutionStarted { .. }
            | StepEventKind::ProgressReset { .. }
            | StepEventKind::AttemptRetry { .. }
            | StepEventKind::ExecutionFailed { .. }
            | StepEventKind::ExecutionAborted { .. }
            | StepEventKind::Nested { .. }
            | StepEventKind::Unknown => None,
        }
    }
}

/// Whether a [`StepEvent`] is a terminal event.
///
/// Returned by [`StepEventKind::is_terminal`].
///
/// The update engine guarantees that after a terminal event is seen, no further
/// events are seen.
#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum StepEventIsTerminal {
    /// This is not a terminal event.
    NonTerminal,

    /// This is a terminal event.
    Terminal {
        /// True if execution completed successfully.
        success: bool,
    },
}

/// The priority of a [`StepEvent`].
///
/// Returned by [`StepEventKind::priority`].
///
/// Some [`StepEvent`]s have a higher priority than others. For example, events
/// related to step successes and failures must be delivered, while events
/// related to retries can be trimmed down since they are overall less
/// important.
///
/// More precisely, a high-priority event is an event which cannot be dropped if
/// an [`EventBuffer`](crate::buffer::EventBuffer) is to work correctly. Low-priority
/// events can be dropped.
#[derive(Clone, Copy, Debug, Eq, PartialEq, PartialOrd, Ord)]
pub enum StepEventPriority {
    /// A low-priority event.
    ///
    /// Includes retry, reset, and unknown events.
    Low,

    /// A high-priority event.
    ///
    /// Includes successes, failures, and terminal events.
    High,
}

#[derive(Deserialize, Serialize)]
#[cfg_attr(feature = "schemars08", derive(schemars::JsonSchema))]
#[derive_where(Clone, Debug, Eq, PartialEq)]
#[serde(bound = "", rename_all = "snake_case", tag = "kind")]
#[cfg_attr(
    feature = "schemars08",
    schemars(rename = "StepOutcomeFor{S}", bound = "S: JsonSchemaEngineSpec",)
)]
pub enum StepOutcome<S: EngineSpec> {
    /// The step completed successfully.
    Success {
        /// An optional message associated with this step.
        message: Option<Cow<'static, str>>,

        /// Optional completion metadata associated with the step.
        metadata: Option<S::CompletionMetadata>,
    },

    /// The step completed with a warning.
    Warning {
        /// A warning message.
        message: Cow<'static, str>,

        /// Optional completion metadata associated with the step.
        metadata: Option<S::CompletionMetadata>,
    },

    /// The step was skipped with a message.
    Skipped {
        /// Message associated with the skip.
        message: Cow<'static, str>,

        /// Optional metadata associated with the skip.
        metadata: Option<S::SkippedMetadata>,
    },
}

impl<S: EngineSpec> StepOutcome<S> {
    /// Converts a generic version into self.
    ///
    /// This version can be used to convert a generic type into a more concrete
    /// form.
    pub fn from_generic(
        value: StepOutcome<GenericSpec>,
    ) -> Result<Self, ConvertGenericError> {
        let ret = match value {
            StepOutcome::Success { message, metadata } => {
                StepOutcome::Success {
                    message,
                    metadata: metadata
                        .map(|metadata| {
                            serde_json::from_value(metadata).map_err(|error| {
                                ConvertGenericError::new("metadata", error)
                            })
                        })
                        .transpose()?,
                }
            }
            StepOutcome::Warning { message, metadata } => {
                StepOutcome::Warning {
                    message,
                    metadata: metadata
                        .map(|metadata| {
                            serde_json::from_value(metadata).map_err(|error| {
                                ConvertGenericError::new("metadata", error)
                            })
                        })
                        .transpose()?,
                }
            }
            StepOutcome::Skipped { message, metadata } => {
                StepOutcome::Skipped {
                    message,
                    metadata: metadata
                        .map(|metadata| {
                            serde_json::from_value(metadata).map_err(|error| {
                                ConvertGenericError::new("metadata", error)
                            })
                        })
                        .transpose()?,
                }
            }
        };
        Ok(ret)
    }

    /// If this outcome represents completion, returns the metadata associated
    /// with this event.
    ///
    /// Returns `None` if this outcome represents "skipped".
    pub fn completion_metadata(&self) -> Option<&S::CompletionMetadata> {
        match self {
            StepOutcome::Success { metadata, .. }
            | StepOutcome::Warning { metadata, .. } => metadata.as_ref(),
            StepOutcome::Skipped { .. } => None,
        }
    }

    /// Converts self into its generic version.
    ///
    /// This version can be used to share data across different kinds of
    /// engines.
    ///
    /// If any of the data in self fails to serialize to a
    /// [`serde_json::Value`], it will be replaced with
    /// [`serde_json::Value::Null`]. Since `serde_json::Value` represents
    /// an arbitrary JSON value, such data would have failed to serialize
    /// anyway.
    pub fn into_generic(self) -> StepOutcome<GenericSpec> {
        match self {
            StepOutcome::Success { message, metadata } => {
                StepOutcome::Success {
                    message,
                    metadata: metadata.map(|metadata| {
                        serde_json::to_value(metadata)
                            .unwrap_or(serde_json::Value::Null)
                    }),
                }
            }
            StepOutcome::Warning { metadata, message } => {
                StepOutcome::Warning {
                    message,
                    metadata: metadata.map(|metadata| {
                        serde_json::to_value(metadata)
                            .unwrap_or(serde_json::Value::Null)
                    }),
                }
            }
            StepOutcome::Skipped { metadata, message } => {
                StepOutcome::Skipped {
                    message,
                    metadata: metadata.map(|metadata| {
                        serde_json::to_value(metadata)
                            .unwrap_or(serde_json::Value::Null)
                    }),
                }
            }
        }
    }

    /// Returns true if the step was successful, including success with
    /// warning.
    pub fn is_success_or_warning(&self) -> bool {
        match self {
            Self::Success { .. } | Self::Warning { .. } => true,
            Self::Skipped { .. } => false,
        }
    }

    /// Returns true if the step was skipped.
    pub fn is_skipped(&self) -> bool {
        match self {
            Self::Skipped { .. } => true,
            Self::Success { .. } | Self::Warning { .. } => false,
        }
    }

    /// Returns the message associated with this outcome, if one exists.
    pub fn message(&self) -> Option<&Cow<'static, str>> {
        match self {
            StepOutcome::Success { message, .. } => message.as_ref(),
            StepOutcome::Warning { message, .. }
            | StepOutcome::Skipped { message, .. } => Some(message),
        }
    }
}

#[derive(Deserialize, Serialize)]
#[derive_where(Clone, Debug, Eq, PartialEq)]
#[serde(bound = "", rename_all = "snake_case")]
pub struct ProgressEvent<S: EngineSpec> {
    /// The specification that this event belongs to.
    ///
    /// This is typically the name of the type `S` for which `EngineSpec` is
    /// implemented.
    ///
    /// This can be used with `Self::from_generic` to deserialize generic
    /// metadata.
    pub spec: String,

    /// The execution ID.
    pub execution_id: ExecutionUuid,

    /// Total time elapsed since the start of execution.
    pub total_elapsed: Duration,

    /// The kind of event this is.
    #[serde(rename = "data")]
    pub kind: ProgressEventKind<S>,
}

#[cfg(feature = "schemars08")]
impl<S: JsonSchemaEngineSpec> schemars::JsonSchema for ProgressEvent<S> {
    fn schema_name() -> String {
        format!("ProgressEventFor{}", S::schema_name())
    }

    fn json_schema(
        generator: &mut schemars::r#gen::SchemaGenerator,
    ) -> schemars::schema::Schema {
        use crate::schema::with_description;
        use schemars::schema::{ObjectValidation, SchemaObject};

        let mut obj = ObjectValidation::default();
        obj.properties.insert(
            "spec".to_owned(),
            with_description(
                generator.subschema_for::<String>(),
                "The specification that this event belongs to.",
            ),
        );
        obj.properties.insert(
            "execution_id".to_owned(),
            with_description(
                generator.subschema_for::<ExecutionUuid>(),
                "The execution ID.",
            ),
        );
        obj.properties.insert(
            "total_elapsed".to_owned(),
            with_description(
                generator.subschema_for::<Duration>(),
                "Total time elapsed since the start of execution.",
            ),
        );
        obj.properties.insert(
            "data".to_owned(),
            with_description(
                generator.subschema_for::<ProgressEventKind<S>>(),
                "The kind of event this is.",
            ),
        );
        obj.required = ["spec", "execution_id", "total_elapsed", "data"]
            .into_iter()
            .map(String::from)
            .collect();

        let mut extensions = serde_json::Map::new();
        if let Some(info) = S::rust_type_info() {
            extensions.insert(
                "x-rust-type".to_owned(),
                crate::schema::rust_type_for_generic(&info, "ProgressEvent"),
            );
        }

        SchemaObject {
            instance_type: Some(schemars::schema::InstanceType::Object.into()),
            object: Some(Box::new(obj)),
            extensions: extensions.into_iter().collect(),
            ..Default::default()
        }
        .into()
    }
}

impl<S: EngineSpec> ProgressEvent<S> {
    /// Converts a generic version into self.
    ///
    /// This version can be used to convert a generic type into a more concrete
    /// form.
    pub fn from_generic(
        value: ProgressEvent<GenericSpec>,
    ) -> Result<Self, ConvertGenericError> {
        Ok(Self {
            spec: value.spec,
            execution_id: value.execution_id,
            total_elapsed: value.total_elapsed,
            kind: ProgressEventKind::from_generic(value.kind)
                .map_err(|error| error.parent("kind"))?,
        })
    }

    /// Converts self into its generic version.
    ///
    /// This version can be used to share data across different kinds of
    /// engines.
    ///
    /// If any of the data in self fails to serialize to a
    /// [`serde_json::Value`], it will be replaced with
    /// [`serde_json::Value::Null`]. Since `serde_json::Value` represents
    /// an arbitrary JSON value, such data would have failed to serialize
    /// anyway.
    pub fn into_generic(self) -> ProgressEvent<GenericSpec> {
        ProgressEvent {
            spec: self.spec,
            execution_id: self.execution_id,
            total_elapsed: self.total_elapsed,
            kind: self.kind.into_generic(),
        }
    }
}

#[derive(Deserialize, Serialize)]
#[cfg_attr(feature = "schemars08", derive(schemars::JsonSchema))]
#[derive_where(Clone, Debug, Eq, PartialEq)]
#[serde(bound = "", rename_all = "snake_case", tag = "kind")]
#[cfg_attr(
    feature = "schemars08",
    schemars(
        rename = "ProgressEventKindFor{S}",
        bound = "S: JsonSchemaEngineSpec",
    )
)]
pub enum ProgressEventKind<S: EngineSpec> {
    /// The update engine is waiting for a progress message.
    ///
    /// The update engine sends this message immediately after a [`StepEvent`]
    /// corresponding to a new step.
    WaitingForProgress {
        /// Information about the step.
        step: StepInfoWithMetadata<S>,

        /// The attempt number currently being executed.
        attempt: usize,

        /// Total time elapsed since the start of the step. Includes prior
        /// attempts.
        step_elapsed: Duration,

        /// Total time elapsed since the start of the attempt.
        attempt_elapsed: Duration,
    },

    Progress {
        /// Information about the step.
        step: StepInfoWithMetadata<S>,

        /// The attempt number currently being executed.
        attempt: usize,

        /// Metadata that was returned with progress.
        metadata: S::ProgressMetadata,

        /// Current progress.
        progress: Option<ProgressCounter>,

        /// Total time elapsed since the start of the step. Includes prior
        /// attempts.
        step_elapsed: Duration,

        /// Total time elapsed since the start of the attempt.
        attempt_elapsed: Duration,
    },

    Nested {
        /// Information about the step.
        step: StepInfoWithMetadata<S>,

        /// The attempt number currently being executed.
        attempt: usize,

        /// The event that occurred.
        event: Box<ProgressEvent<GenericSpec>>,

        /// Total time elapsed since the start of the step. Includes prior
        /// attempts.
        step_elapsed: Duration,

        /// The time it took for this attempt to complete.
        attempt_elapsed: Duration,
    },

    /// Future variants that might be unknown.
    #[serde(other, deserialize_with = "deserialize_ignore_any")]
    Unknown,
}

/// The progress of a leaf event.
///
/// Returned by [`ProgressEventKind::leaf_progress`].
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum LeafProgress<'a> {
    WaitingForProgress,
    Progress(Option<&'a ProgressCounter>),
    Unknown,
}

impl<S: EngineSpec> ProgressEventKind<S> {
    /// Returns the progress counter for this event, if available.
    pub fn progress_counter(&self) -> Option<&ProgressCounter> {
        match self.leaf_progress() {
            LeafProgress::Progress(counter) => counter,
            LeafProgress::WaitingForProgress | LeafProgress::Unknown => None,
        }
    }

    /// Returns a [`LeafProgress`] for this event.
    ///
    /// This is similar to [`Self::progress_counter`], but distinguishes
    /// "waiting for progress" from "unknown progress".
    pub fn leaf_progress(&self) -> LeafProgress<'_> {
        match self {
            ProgressEventKind::WaitingForProgress { .. } => {
                LeafProgress::WaitingForProgress
            }
            ProgressEventKind::Progress { progress, .. } => {
                LeafProgress::Progress(progress.as_ref())
            }
            ProgressEventKind::Nested { event, .. } => {
                event.kind.leaf_progress()
            }
            ProgressEventKind::Unknown => LeafProgress::Unknown,
        }
    }

    /// Returns `attempt` for the leaf event, recursing into nested events as
    /// necessary.
    ///
    /// Returns None for unknown events.
    pub fn leaf_attempt(&self) -> Option<usize> {
        match self {
            ProgressEventKind::WaitingForProgress { attempt, .. }
            | ProgressEventKind::Progress { attempt, .. } => Some(*attempt),
            ProgressEventKind::Nested { event, .. } => {
                event.kind.leaf_attempt()
            }
            ProgressEventKind::Unknown => None,
        }
    }

    /// Returns `step_elapsed` for the leaf event, recursing into nested events
    /// as necessary.
    ///
    /// Returns None for unknown events.
    pub fn leaf_step_elapsed(&self) -> Option<Duration> {
        match self {
            ProgressEventKind::WaitingForProgress { step_elapsed, .. }
            | ProgressEventKind::Progress { step_elapsed, .. } => {
                Some(*step_elapsed)
            }
            ProgressEventKind::Nested { event, .. } => {
                event.kind.leaf_step_elapsed()
            }
            ProgressEventKind::Unknown => None,
        }
    }

    /// Returns `attempt_elapsed` for the leaf event, recursing into nested
    /// events as necessary.
    ///
    /// Returns None for unknown events.
    pub fn leaf_attempt_elapsed(&self) -> Option<Duration> {
        match self {
            ProgressEventKind::WaitingForProgress {
                attempt_elapsed, ..
            }
            | ProgressEventKind::Progress { attempt_elapsed, .. } => {
                Some(*attempt_elapsed)
            }
            ProgressEventKind::Nested { event, .. } => {
                event.kind.leaf_attempt_elapsed()
            }
            ProgressEventKind::Unknown => None,
        }
    }

    /// Converts a generic version into self.
    ///
    /// This version can be used to convert a generic type into a more concrete
    /// form.
    pub fn from_generic(
        value: ProgressEventKind<GenericSpec>,
    ) -> Result<Self, ConvertGenericError> {
        let ret = match value {
            ProgressEventKind::WaitingForProgress {
                step,
                attempt,
                step_elapsed,
                attempt_elapsed,
            } => ProgressEventKind::WaitingForProgress {
                step: StepInfoWithMetadata::from_generic(step)
                    .map_err(|error| error.parent("step"))?,
                attempt,
                step_elapsed,
                attempt_elapsed,
            },
            ProgressEventKind::Progress {
                step,
                attempt,
                metadata,
                progress,
                step_elapsed,
                attempt_elapsed,
            } => ProgressEventKind::Progress {
                step: StepInfoWithMetadata::from_generic(step)
                    .map_err(|error| error.parent("step"))?,
                attempt,
                metadata: serde_json::from_value(metadata).map_err(
                    |error| ConvertGenericError::new("metadata", error),
                )?,
                progress,
                step_elapsed,
                attempt_elapsed,
            },
            ProgressEventKind::Nested {
                step,
                attempt,
                event,
                step_elapsed,
                attempt_elapsed,
            } => ProgressEventKind::Nested {
                step: StepInfoWithMetadata::from_generic(step)
                    .map_err(|error| error.parent("step"))?,
                attempt,
                event,
                step_elapsed,
                attempt_elapsed,
            },
            ProgressEventKind::Unknown => ProgressEventKind::Unknown,
        };
        Ok(ret)
    }

    /// Converts self into its generic version.
    ///
    /// This version can be used to share data across different kinds of
    /// engines.
    ///
    /// If any of the data in self fails to serialize to a
    /// [`serde_json::Value`], it will be replaced with
    /// [`serde_json::Value::Null`]. Since `serde_json::Value` represents
    /// an arbitrary JSON value, such data would have failed to serialize
    /// anyway.
    pub fn into_generic(self) -> ProgressEventKind<GenericSpec> {
        match self {
            ProgressEventKind::WaitingForProgress {
                step,
                attempt,
                step_elapsed,
                attempt_elapsed,
            } => ProgressEventKind::WaitingForProgress {
                step: step.into_generic(),
                attempt,
                step_elapsed,
                attempt_elapsed,
            },
            ProgressEventKind::Progress {
                step,
                attempt,
                metadata,
                progress,
                step_elapsed,
                attempt_elapsed,
            } => ProgressEventKind::Progress {
                step: step.into_generic(),
                attempt,
                metadata: serde_json::to_value(metadata)
                    .unwrap_or(serde_json::Value::Null),
                progress,
                step_elapsed,
                attempt_elapsed,
            },
            ProgressEventKind::Nested {
                step,
                attempt,
                event,
                step_elapsed,
                attempt_elapsed,
            } => ProgressEventKind::Nested {
                step: step.into_generic(),
                attempt,
                event,
                step_elapsed,
                attempt_elapsed,
            },
            ProgressEventKind::Unknown => ProgressEventKind::Unknown,
        }
    }
}

/// Serializable information about a step.
#[derive(Deserialize, Serialize)]
#[cfg_attr(feature = "schemars08", derive(schemars::JsonSchema))]
#[derive_where(Clone, Debug, Eq, PartialEq)]
#[serde(bound = "", rename_all = "snake_case")]
#[cfg_attr(
    feature = "schemars08",
    schemars(rename = "StepInfoFor{S}", bound = "S: JsonSchemaEngineSpec",)
)]
pub struct StepInfo<S: EngineSpec> {
    /// An identifier for this step.
    pub id: S::StepId,

    /// The component that this step is part of.
    pub component: S::Component,

    /// The description for this step.
    pub description: Cow<'static, str>,

    /// The index of the step within all steps to be executed.
    pub index: usize,

    /// The index of the step within the component.
    pub component_index: usize,

    /// The total number of steps in this component.
    pub total_component_steps: usize,
}

impl<S: EngineSpec> StepInfo<S> {
    /// Returns true if this is the last step in this component.
    pub fn is_last_step_in_component(&self) -> bool {
        self.component_index + 1 == self.total_component_steps
    }

    /// Converts a generic version into self.
    ///
    /// This version can be used to convert a generic type into a more concrete
    /// form.
    pub fn from_generic(
        value: StepInfo<GenericSpec>,
    ) -> Result<Self, ConvertGenericError> {
        Ok(Self {
            id: serde_json::from_value(value.id)
                .map_err(|error| ConvertGenericError::new("id", error))?,
            component: serde_json::from_value(value.component).map_err(
                |error| ConvertGenericError::new("component", error),
            )?,
            description: value.description,
            index: value.index,
            component_index: value.component_index,
            total_component_steps: value.total_component_steps,
        })
    }

    /// Converts self into its generic version.
    ///
    /// This version can be used to share data across different kinds of
    /// engines.
    ///
    /// If any of the data in self fails to serialize to a
    /// [`serde_json::Value`], it will be replaced with
    /// [`serde_json::Value::Null`]. Since `serde_json::Value` represents
    /// an arbitrary JSON value, such data would have failed to serialize
    /// anyway.
    pub fn into_generic(self) -> StepInfo<GenericSpec> {
        StepInfo {
            id: serde_json::to_value(self.id)
                .unwrap_or(serde_json::Value::Null),
            component: serde_json::to_value(self.component)
                .unwrap_or(serde_json::Value::Null),
            description: self.description,
            index: self.index,
            component_index: self.component_index,
            total_component_steps: self.total_component_steps,
        }
    }
}

#[derive(Deserialize, Serialize)]
#[cfg_attr(feature = "schemars08", derive(schemars::JsonSchema))]
#[derive_where(Clone, Debug, Eq, PartialEq)]
#[serde(bound = "", rename_all = "snake_case")]
#[cfg_attr(
    feature = "schemars08",
    schemars(
        rename = "StepComponentSummaryFor{S}",
        bound = "S: JsonSchemaEngineSpec",
    )
)]
pub struct StepComponentSummary<S: EngineSpec> {
    /// The component.
    pub component: S::Component,

    /// The number of steps present in this component.
    pub total_component_steps: usize,
}

impl<S: EngineSpec> StepComponentSummary<S> {
    /// Converts a generic version into self.
    ///
    /// This version can be used to convert a generic type into a more concrete
    /// form.
    pub fn from_generic(
        value: StepComponentSummary<GenericSpec>,
    ) -> Result<Self, ConvertGenericError> {
        Ok(Self {
            component: serde_json::from_value(value.component).map_err(
                |error| ConvertGenericError::new("component", error),
            )?,
            total_component_steps: value.total_component_steps,
        })
    }

    /// Converts self into its generic version.
    ///
    /// This version can be used to share data across different kinds of
    /// engines.
    ///
    /// If any of the data in self fails to serialize to a
    /// [`serde_json::Value`], it will be replaced with
    /// [`serde_json::Value::Null`]. Since `serde_json::Value` represents
    /// an arbitrary JSON value, such data would have failed to serialize
    /// anyway.
    pub fn into_generic(self) -> StepComponentSummary<GenericSpec> {
        StepComponentSummary {
            component: serde_json::to_value(self.component)
                .unwrap_or(serde_json::Value::Null),
            total_component_steps: self.total_component_steps,
        }
    }
}

/// Serializable information about a step.
#[derive(Deserialize, Serialize)]
#[cfg_attr(feature = "schemars08", derive(schemars::JsonSchema))]
#[derive_where(Clone, Debug, Eq, PartialEq)]
#[serde(bound = "", rename_all = "snake_case")]
#[cfg_attr(
    feature = "schemars08",
    schemars(
        rename = "StepInfoWithMetadataFor{S}",
        bound = "S: JsonSchemaEngineSpec",
    )
)]
pub struct StepInfoWithMetadata<S: EngineSpec> {
    /// Information about this step.
    pub info: StepInfo<S>,

    /// Additional metadata associated with this step.
    pub metadata: Option<S::StepMetadata>,
}

impl<S: EngineSpec> StepInfoWithMetadata<S> {
    /// Converts a generic version into self.
    ///
    /// This version can be used to convert a generic type into a more concrete
    /// form.
    pub fn from_generic(
        value: StepInfoWithMetadata<GenericSpec>,
    ) -> Result<Self, ConvertGenericError> {
        Ok(Self {
            info: StepInfo::from_generic(value.info)
                .map_err(|error| error.parent("info"))?,
            metadata: value
                .metadata
                .map(|metadata| {
                    serde_json::from_value(metadata).map_err(|error| {
                        ConvertGenericError::new("metadata", error)
                    })
                })
                .transpose()?,
        })
    }

    /// Converts self into its generic version.
    ///
    /// This version can be used to share data across different kinds of
    /// engines.
    ///
    /// If any of the data in self fails to serialize to a
    /// [`serde_json::Value`], it will be replaced with
    /// [`serde_json::Value::Null`]. Since `serde_json::Value` represents
    /// an arbitrary JSON value, such data would have failed to serialize
    /// anyway.
    pub fn into_generic(self) -> StepInfoWithMetadata<GenericSpec> {
        StepInfoWithMetadata {
            info: self.info.into_generic(),
            metadata: self.metadata.map(|metadata| {
                serde_json::to_value(metadata)
                    .unwrap_or(serde_json::Value::Null)
            }),
        }
    }
}

/// Current progress.
///
/// Both `current` and `total` are abstract counters. These counters are often a
/// number of bytes. There is no guarantee that the counter won't go back in
/// subsequent events; that can happen e.g. if a fetch happens from multiple
/// peers within a single attempt.
#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
#[serde(rename_all = "snake_case")]
pub struct ProgressCounter {
    /// The current progress.
    pub current: u64,

    /// The total progress.
    pub total: Option<u64>,

    /// Progress units.
    pub units: ProgressUnits,
}

#[cfg(feature = "schemars08")]
impl schemars::JsonSchema for ProgressCounter {
    fn schema_name() -> String {
        "ProgressCounter".to_owned()
    }

    fn json_schema(
        generator: &mut schemars::r#gen::SchemaGenerator,
    ) -> schemars::schema::Schema {
        use crate::schema::{
            EVENTS_MODULE, rust_type_for_events, with_description,
        };
        use schemars::schema::{ObjectValidation, SchemaObject};

        let mut obj = ObjectValidation::default();
        obj.properties.insert(
            "current".to_owned(),
            with_description(
                generator.subschema_for::<u64>(),
                "The current progress.",
            ),
        );
        obj.properties.insert(
            "total".to_owned(),
            with_description(
                generator.subschema_for::<Option<u64>>(),
                "The total progress.",
            ),
        );
        obj.properties.insert(
            "units".to_owned(),
            with_description(
                generator.subschema_for::<ProgressUnits>(),
                "Progress units.",
            ),
        );
        obj.required =
            ["current", "units"].into_iter().map(String::from).collect();

        SchemaObject {
            metadata: Some(Box::new(schemars::schema::Metadata {
                description: Some("Current progress.".to_owned()),
                ..Default::default()
            })),
            instance_type: Some(schemars::schema::InstanceType::Object.into()),
            object: Some(Box::new(obj)),
            extensions: [(
                "x-rust-type".to_owned(),
                rust_type_for_events(&format!(
                    "{EVENTS_MODULE}::ProgressCounter"
                )),
            )]
            .into_iter()
            .collect(),
            ..Default::default()
        }
        .into()
    }
}

impl ProgressCounter {
    /// Creates a new `ProgressCounter` with current and total values.
    #[inline]
    pub fn new(
        current: u64,
        total: u64,
        units: impl Into<ProgressUnits>,
    ) -> Self {
        Self { current, total: Some(total), units: units.into() }
    }

    /// Creates a new `ProgressCounter` with just a current value.
    #[inline]
    pub fn current(current: u64, units: impl Into<ProgressUnits>) -> Self {
        Self { current, total: None, units: units.into() }
    }
}

#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
#[serde(transparent)]
pub struct ProgressUnits(pub Cow<'static, str>);

#[cfg(feature = "schemars08")]
impl schemars::JsonSchema for ProgressUnits {
    fn schema_name() -> String {
        "ProgressUnits".to_owned()
    }

    fn json_schema(
        generator: &mut schemars::r#gen::SchemaGenerator,
    ) -> schemars::schema::Schema {
        use crate::schema::{EVENTS_MODULE, rust_type_for_events};

        // ProgressUnits is a transparent wrapper around a string.
        // Delegate to String's schema and add x-rust-type.
        let mut schema = match generator.subschema_for::<String>() {
            schemars::schema::Schema::Object(obj) => obj,
            // String always produces an Object schema with
            // schemars 0.8. If this changes, we need to handle
            // it explicitly.
            other => {
                debug_assert!(
                    false,
                    "expected String to produce an Object \
                     schema, got: {other:?}",
                );
                return other;
            }
        };
        schema.extensions.insert(
            "x-rust-type".to_owned(),
            rust_type_for_events(&format!("{EVENTS_MODULE}::ProgressUnits")),
        );
        schema.into()
    }
}

impl ProgressUnits {
    /// Creates a new `ProgressUnits`.
    pub fn new(s: impl Into<Cow<'static, str>>) -> Self {
        Self(s.into())
    }

    /// Creates a new `ProgressUnits` from a static string.
    pub const fn new_const(s: &'static str) -> Self {
        Self(Cow::Borrowed(s))
    }

    /// Units in terms of bytes.
    ///
    /// Some displayers might display bytes in terms of KiB, MiB etc.
    pub const BYTES: Self = Self::new_const("bytes");
}

impl AsRef<str> for ProgressUnits {
    fn as_ref(&self) -> &str {
        self.0.as_ref()
    }
}

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

impl<T> From<T> for ProgressUnits
where
    T: Into<Cow<'static, str>>,
{
    fn from(value: T) -> Self {
        Self(value.into())
    }
}

#[derive_where(Clone, Debug, Eq, PartialEq)]
pub enum StepProgress<S: EngineSpec> {
    /// A step has progressed.
    Progress {
        /// Current progress.
        progress: Option<ProgressCounter>,

        /// Metadata associated with progress.
        metadata: S::ProgressMetadata,
    },

    /// Progress was reset: typically, the step failed along one path, and the
    /// step is now trying a different path.
    ///
    /// For example, downloading from one peer failed and we've moved to the
    /// next peer.
    Reset {
        /// Metadata associated with the reset.
        metadata: S::ProgressMetadata,

        /// A message associated with the reset.
        message: Cow<'static, str>,
    },

    /// The step is being retried from the beginning.
    Retry {
        /// An optional message associated with the retry.
        message: Cow<'static, str>,
    },
}

impl<S: EngineSpec> StepProgress<S> {
    /// Creates a new progress message with current and total values.
    pub fn with_current_and_total(
        current: u64,
        total: u64,
        units: impl Into<ProgressUnits>,
        metadata: S::ProgressMetadata,
    ) -> Self {
        Self::Progress {
            progress: Some(ProgressCounter {
                current,
                total: Some(total),
                units: units.into(),
            }),
            metadata,
        }
    }

    /// Creates a new progress message with a current value.
    pub fn with_current(
        current: u64,
        units: impl Into<ProgressUnits>,
        metadata: S::ProgressMetadata,
    ) -> Self {
        Self::Progress {
            progress: Some(ProgressCounter {
                current,
                total: None,
                units: units.into(),
            }),
            metadata,
        }
    }

    /// Creates a new progress message without either current or total values.
    pub fn progress(metadata: S::ProgressMetadata) -> Self {
        Self::Progress { progress: None, metadata }
    }

    /// Creates a new reset message.
    pub fn reset(
        metadata: S::ProgressMetadata,
        message: impl Into<Cow<'static, str>>,
    ) -> Self {
        Self::Reset { metadata, message: message.into() }
    }

    /// Creates a new retry message.
    pub fn retry(message: impl Into<Cow<'static, str>>) -> Self {
        Self::Retry { message: message.into() }
    }
}

/// A report produced from an [`EventBuffer`](crate::buffer::EventBuffer).
///
/// Remote reports can be passed into a `StepContext`, in which case
/// they show up as nested events.
#[derive_where(Clone, Debug, Default, Eq, PartialEq)]
#[derive(Deserialize, Serialize)]
#[serde(bound = "", rename_all = "snake_case")]
pub struct EventReport<S: EngineSpec> {
    /// A list of step events.
    ///
    /// Step events include success and failure events.
    pub step_events: Vec<StepEvent<S>>,

    /// A list of progress events, or whether we're currently waiting for a
    /// progress event.
    ///
    /// Currently, this produces one progress event for each top-level and
    /// nested event in progress.
    pub progress_events: Vec<ProgressEvent<S>>,

    /// The root execution ID for this report.
    ///
    /// Each report has a root execution ID, which ties together all step and
    /// progress events. This is always filled out if the list of step events is
    /// non-empty.
    pub root_execution_id: Option<ExecutionUuid>,

    /// The last event seen.
    ///
    /// `last_seen` can be used to retrieve deltas of events.
    pub last_seen: Option<usize>,
}

#[cfg(feature = "schemars08")]
impl<S: JsonSchemaEngineSpec> schemars::JsonSchema for EventReport<S> {
    fn schema_name() -> String {
        format!("EventReportFor{}", S::schema_name())
    }

    fn json_schema(
        generator: &mut schemars::r#gen::SchemaGenerator,
    ) -> schemars::schema::Schema {
        use crate::schema::with_description;
        use schemars::schema::{ObjectValidation, SchemaObject};

        let mut obj = ObjectValidation::default();
        obj.properties.insert(
            "step_events".to_owned(),
            with_description(
                generator.subschema_for::<Vec<StepEvent<S>>>(),
                "A list of step events.",
            ),
        );
        obj.properties.insert(
            "progress_events".to_owned(),
            with_description(
                generator.subschema_for::<Vec<ProgressEvent<S>>>(),
                "A list of progress events, or whether we're \
                 currently waiting for a progress event.",
            ),
        );
        obj.properties.insert(
            "root_execution_id".to_owned(),
            with_description(
                generator.subschema_for::<Option<ExecutionUuid>>(),
                "The root execution ID for this report.",
            ),
        );
        obj.properties.insert(
            "last_seen".to_owned(),
            with_description(
                generator.subschema_for::<Option<usize>>(),
                "The last event seen.",
            ),
        );
        obj.required = ["step_events", "progress_events"]
            .into_iter()
            .map(String::from)
            .collect();

        let mut extensions = serde_json::Map::new();
        if let Some(info) = S::rust_type_info() {
            extensions.insert(
                "x-rust-type".to_owned(),
                crate::schema::rust_type_for_generic(&info, "EventReport"),
            );
        }

        SchemaObject {
            metadata: Some(Box::new(schemars::schema::Metadata {
                description: Some(
                    "An oxide-update-engine event report.\
                     \n\nRemote reports can be passed into a \
                     `StepContext`, in which case they show up as \
                     nested events."
                        .to_owned(),
                ),
                ..Default::default()
            })),
            instance_type: Some(schemars::schema::InstanceType::Object.into()),
            object: Some(Box::new(obj)),
            extensions: extensions.into_iter().collect(),
            ..Default::default()
        }
        .into()
    }
}

impl<S: EngineSpec> EventReport<S> {
    /// Converts a generic version into self.
    ///
    /// This version can be used to convert a generic type into a more concrete
    /// form.
    pub fn from_generic(
        value: EventReport<GenericSpec>,
    ) -> Result<Self, ConvertGenericError> {
        Ok(Self {
            step_events: value
                .step_events
                .into_iter()
                .enumerate()
                .map(|(index, event)| {
                    StepEvent::from_generic(event).map_err(|error| {
                        error.parent_array("step_events", index)
                    })
                })
                .collect::<Result<Vec<_>, _>>()?,
            progress_events: value
                .progress_events
                .into_iter()
                .enumerate()
                .map(|(index, event)| {
                    ProgressEvent::from_generic(event).map_err(|error| {
                        error.parent_array("progress_events", index)
                    })
                })
                .collect::<Result<Vec<_>, _>>()?,
            root_execution_id: value.root_execution_id,
            last_seen: value.last_seen,
        })
    }

    /// Converts self into its generic version.
    ///
    /// This version can be used to share data across different kinds of
    /// engines.
    ///
    /// If any of the data in self fails to serialize to a
    /// [`serde_json::Value`], it will be replaced with
    /// [`serde_json::Value::Null`]. Since `serde_json::Value` represents
    /// an arbitrary JSON value, such data would have failed to serialize
    /// anyway.
    pub fn into_generic(self) -> EventReport<GenericSpec> {
        EventReport {
            step_events: self
                .step_events
                .into_iter()
                .map(|event| event.into_generic())
                .collect(),
            progress_events: self
                .progress_events
                .into_iter()
                .map(|event| event.into_generic())
                .collect(),
            root_execution_id: self.root_execution_id,
            last_seen: self.last_seen,
        }
    }
}

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

    enum TestSpec {}

    impl EngineSpec for TestSpec {
        fn spec_name() -> String {
            "TestSpec".to_owned()
        }

        type Component = String;
        type StepId = usize;
        type StepMetadata = serde_json::Value;
        type ProgressMetadata = serde_json::Value;
        type CompletionMetadata = serde_json::Value;
        type SkippedMetadata = serde_json::Value;
        type Error = anyhow::Error;
    }

    fn test_execution_id() -> ExecutionUuid {
        "2cc08a14-5e96-4917-bc70-e98293a3b703".parse().expect("parsed UUID")
    }

    #[test]
    fn step_event_parse_unknown() {
        let execution_id = test_execution_id();
        let tests = [
            (
                r#"
                  {
                    "spec": "TestSpec",
                    "execution_id": "2cc08a14-5e96-4917-bc70-e98293a3b703",
                    "event_index": 0,
                    "total_elapsed": {
                      "secs": 0,
                      "nanos": 0
                    },
                    "data": {
                      "kind": "unknown_variant",
                      "last_step": {
                        "info": {
                          "id": 0,
                          "component": "foo",
                          "description": "Description",
                          "index": 0,
                          "component_index": 0,
                          "total_component_steps": 1
                        },
                        "metadata": null
                      },
                      "last_attempt": 1,
                      "last_outcome": {
                        "kind": "success",
                        "metadata": null
                      },
                      "step_elapsed": {
                        "secs": 0,
                        "nanos": 0
                      },
                      "attempt_elapsed": {
                        "secs": 0,
                        "nanos": 0
                      }
                    }
                  }
                "#,
                StepEvent {
                    spec: TestSpec::spec_name(),
                    execution_id,
                    event_index: 0,
                    total_elapsed: Duration::ZERO,
                    kind: StepEventKind::Unknown,
                },
            ),
            (
                r#"
                  {
                    "spec": "TestSpec",
                    "execution_id": "2cc08a14-5e96-4917-bc70-e98293a3b703",
                    "event_index": 1,
                    "total_elapsed": {
                      "secs": 0,
                      "nanos": 0
                    },
                    "data": {
                      "kind": "execution_completed",
                      "last_step": {
                        "info": {
                          "id": 0,
                          "component": "foo",
                          "description": "Description",
                          "index": 0,
                          "component_index": 0,
                          "total_component_steps": 1
                        },
                        "metadata": null
                      },
                      "last_attempt": 1,
                      "last_outcome": {
                        "kind": "success",
                        "message": null,
                        "metadata": null
                      },
                      "step_elapsed": {
                        "secs": 0,
                        "nanos": 0
                      },
                      "attempt_elapsed": {
                        "secs": 0,
                        "nanos": 0
                      },
                      "unknown_field": 123
                    }
                  }
                "#,
                StepEvent::<TestSpec> {
                    spec: TestSpec::spec_name(),
                    execution_id,
                    event_index: 1,
                    total_elapsed: Duration::ZERO,
                    kind: StepEventKind::ExecutionCompleted {
                        last_step: StepInfoWithMetadata {
                            info: StepInfo {
                                id: 0,
                                component: "foo".to_owned(),
                                description: "Description".into(),
                                index: 0,
                                component_index: 0,
                                total_component_steps: 1,
                            },
                            metadata: None,
                        },
                        last_attempt: 1,
                        last_outcome: StepOutcome::Success {
                            message: None,
                            metadata: None,
                        },
                        step_elapsed: Duration::ZERO,
                        attempt_elapsed: Duration::ZERO,
                    },
                },
            ),
        ];

        for (index, (input, expected)) in tests.into_iter().enumerate() {
            let actual: StepEvent<TestSpec> = serde_json::from_str(input)
                .unwrap_or_else(|error| {
                    panic!("index {index}: unknown variant deserialized correctly: {error}")
                });
            assert_eq!(expected, actual, "input matches actual output");
        }
    }

    #[test]
    fn progress_event_parse_unknown() {
        let execution_id = test_execution_id();

        let tests = [
            (
                r#"
                  {
                    "spec": "TestSpec",
                    "execution_id": "2cc08a14-5e96-4917-bc70-e98293a3b703",
                    "total_elapsed": {
                      "secs": 0,
                      "nanos": 0
                    },
                    "data": {
                      "kind": "unknown_variant",
                      "step": {
                        "info": {
                          "id": 0,
                          "component": "foo",
                          "description": "Description",
                          "index": 0,
                          "component_index": 0,
                          "total_component_steps": 1
                        },
                        "metadata": null
                      },
                      "attempt": 1,
                      "metadata": null,
                      "progress": {
                        "current": 123,
                        "total": null
                      },
                      "step_elapsed": {
                        "secs": 0,
                        "nanos": 0
                      },
                      "attempt_elapsed": {
                        "secs": 0,
                        "nanos": 0
                      }
                    }
                  }
                "#,
                ProgressEvent {
                    spec: TestSpec::spec_name(),
                    execution_id,
                    total_elapsed: Duration::ZERO,
                    kind: ProgressEventKind::Unknown,
                },
            ),
            (
                r#"
                  {
                    "spec": "TestSpec",
                    "execution_id": "2cc08a14-5e96-4917-bc70-e98293a3b703",
                    "total_elapsed": {
                      "secs": 0,
                      "nanos": 0
                    },
                    "data": {
                      "kind": "progress",
                      "step": {
                        "info": {
                          "id": 0,
                          "component": "foo",
                          "description": "Description",
                          "index": 0,
                          "component_index": 0,
                          "total_component_steps": 1
                        },
                        "metadata": null
                      },
                      "attempt": 1,
                      "metadata": null,
                      "progress": {
                        "current": 123,
                        "total": null,
                        "units": "bytes"
                      },
                      "step_elapsed": {
                        "secs": 0,
                        "nanos": 0
                      },
                      "attempt_elapsed": {
                        "secs": 0,
                        "nanos": 0
                      },
                      "unknown_field": 123
                    }
                  }
                "#,
                ProgressEvent::<TestSpec> {
                    spec: TestSpec::spec_name(),
                    execution_id,
                    total_elapsed: Duration::ZERO,
                    kind: ProgressEventKind::Progress {
                        step: StepInfoWithMetadata {
                            info: StepInfo {
                                id: 0,
                                component: "foo".to_owned(),
                                description: "Description".into(),
                                index: 0,
                                component_index: 0,
                                total_component_steps: 1,
                            },
                            metadata: None,
                        },
                        attempt: 1,
                        metadata: serde_json::Value::Null,
                        progress: Some(ProgressCounter::current(
                            123,
                            ProgressUnits::BYTES,
                        )),
                        step_elapsed: Duration::ZERO,
                        attempt_elapsed: Duration::ZERO,
                    },
                },
            ),
        ];

        for (index, (input, expected)) in tests.into_iter().enumerate() {
            let actual: ProgressEvent<TestSpec> = serde_json::from_str(input)
                .unwrap_or_else(|error| {
                    panic!("index {index}: unknown variant deserialized correctly: {error}")
                });
            assert_eq!(expected, actual, "input matches actual output");
        }
    }

    fn generic_step() -> StepInfoWithMetadata<GenericSpec> {
        StepInfoWithMetadata {
            info: StepInfo {
                id: serde_json::Value::Null,
                component: serde_json::Value::Null,
                description: "Description".into(),
                index: 0,
                component_index: 0,
                total_component_steps: 1,
            },
            metadata: None,
        }
    }

    fn generic_progress_event(
        kind: ProgressEventKind<GenericSpec>,
    ) -> ProgressEvent<GenericSpec> {
        ProgressEvent {
            spec: GenericSpec::spec_name(),
            execution_id: test_execution_id(),
            total_elapsed: Duration::ZERO,
            kind,
        }
    }

    fn generic_progress(
        progress: Option<ProgressCounter>,
    ) -> ProgressEventKind<GenericSpec> {
        ProgressEventKind::Progress {
            step: generic_step(),
            attempt: 1,
            metadata: serde_json::Value::Null,
            progress,
            step_elapsed: Duration::ZERO,
            attempt_elapsed: Duration::ZERO,
        }
    }

    fn generic_waiting() -> ProgressEventKind<GenericSpec> {
        ProgressEventKind::WaitingForProgress {
            step: generic_step(),
            attempt: 1,
            step_elapsed: Duration::ZERO,
            attempt_elapsed: Duration::ZERO,
        }
    }

    fn nest(
        inner: ProgressEventKind<GenericSpec>,
    ) -> ProgressEventKind<GenericSpec> {
        ProgressEventKind::Nested {
            step: generic_step(),
            attempt: 1,
            event: Box::new(generic_progress_event(inner)),
            step_elapsed: Duration::ZERO,
            attempt_elapsed: Duration::ZERO,
        }
    }

    #[test]
    fn leaf_progress_recurses_to_innermost_counter() {
        let counter = ProgressCounter::current(42, ProgressUnits::BYTES);
        let leaf = generic_progress(Some(counter.clone()));

        let nested = nest(nest(leaf));

        assert_eq!(
            nested.leaf_progress(),
            LeafProgress::Progress(Some(&counter)),
            "leaf_progress recurses to the innermost counter",
        );
        assert_eq!(
            nested.progress_counter(),
            Some(&counter),
            "progress_counter agrees with leaf_progress",
        );
    }

    #[test]
    fn leaf_progress_distinguishes_progress_none_from_waiting() {
        let progress_none = generic_progress(None);
        assert_eq!(
            progress_none.leaf_progress(),
            LeafProgress::Progress(None),
            "Progress with no counter maps to Progress(None)",
        );
        assert_eq!(
            progress_none.progress_counter(),
            None,
            "Progress(None) has no progress counter",
        );

        let waiting = generic_waiting();
        assert_eq!(
            waiting.leaf_progress(),
            LeafProgress::WaitingForProgress,
            "WaitingForProgress maps to LeafProgress::WaitingForProgress",
        );
        assert_ne!(
            waiting.leaf_progress(),
            progress_none.leaf_progress(),
            "WaitingForProgress is distinct from Progress(None)",
        );
    }

    #[test]
    fn leaf_progress_recurses_to_nested_waiting() {
        let nested = nest(generic_waiting());
        assert_eq!(
            nested.leaf_progress(),
            LeafProgress::WaitingForProgress,
            "leaf_progress recurses through Nested to WaitingForProgress",
        );
        assert_eq!(
            nested.progress_counter(),
            None,
            "nested WaitingForProgress has no progress counter",
        );
    }

    #[test]
    fn leaf_progress_unknown() {
        let unknown = ProgressEventKind::<GenericSpec>::Unknown;
        assert_eq!(
            unknown.leaf_progress(),
            LeafProgress::Unknown,
            "Unknown maps to LeafProgress::Unknown",
        );
        assert_eq!(
            unknown.progress_counter(),
            None,
            "Unknown has no progress counter",
        );
    }

    // --- Schema tests (schemars08 feature) ---

    #[cfg(feature = "schemars08")]
    mod schema_tests {
        use super::*;
        use crate::schema::RustTypeInfo;
        use schemars::{JsonSchema, r#gen::SchemaGenerator, schema::Schema};

        /// Extract the x-rust-type extension from a schema, if
        /// present.
        fn get_rust_type_ext(schema: &Schema) -> Option<&serde_json::Value> {
            match schema {
                Schema::Object(obj) => obj.extensions.get("x-rust-type"),
                Schema::Bool(_) => None,
            }
        }

        /// Assert that x-rust-type has the expected crate, version,
        /// and path.
        fn assert_rust_type_ext(
            x_rust_type: &serde_json::Value,
            expected_crate: &str,
            expected_version: &str,
            expected_path: &str,
        ) {
            assert_eq!(
                x_rust_type.get("crate").and_then(|v| v.as_str()),
                Some(expected_crate),
                "x-rust-type crate"
            );
            assert_eq!(
                x_rust_type.get("version").and_then(|v| v.as_str()),
                Some(expected_version),
                "x-rust-type version"
            );
            assert_eq!(
                x_rust_type.get("path").and_then(|v| v.as_str()),
                Some(expected_path),
                "x-rust-type path"
            );
        }

        // -- Non-generic types --

        #[test]
        fn execution_uuid_kind_rust_type() {
            let mut generator = SchemaGenerator::default();
            let schema = ExecutionUuidKind::json_schema(&mut generator);
            let xrt = get_rust_type_ext(&schema).expect("x-rust-type present");
            assert_rust_type_ext(
                xrt,
                "oxide-update-engine-types",
                env!("CARGO_PKG_VERSION"),
                "oxide_update_engine_types::events\
                 ::ExecutionUuidKind",
            );
        }

        #[test]
        fn execution_uuid_lifted_rust_type() {
            // The TypedUuid<ExecutionUuidKind> schema should
            // lift the x-rust-type with the alias
            // "ExecutionUuid".
            let mut generator = SchemaGenerator::default();
            let schema = ExecutionUuid::json_schema(&mut generator);
            let xrt = get_rust_type_ext(&schema)
                .expect("x-rust-type present on ExecutionUuid");
            assert_rust_type_ext(
                xrt,
                "oxide-update-engine-types",
                env!("CARGO_PKG_VERSION"),
                "oxide_update_engine_types::events\
                 ::ExecutionUuid",
            );
        }

        // -- Full schema snapshot (covers all types transitively) --

        #[test]
        fn event_report_generic_spec_schema() {
            let schema = schemars::schema_for!(EventReport<GenericSpec>);
            let json = serde_json::to_string_pretty(&schema)
                .expect("serialized schema");
            expectorate::assert_contents(
                "tests/output/event_report_generic_spec_schema.json",
                &json,
            );
        }

        // -- Generic types with a spec that returns None for
        // rust_type_info --

        impl schemars::JsonSchema for super::TestSpec {
            fn schema_name() -> String {
                "TestSpec".to_owned()
            }

            fn json_schema(_: &mut SchemaGenerator) -> Schema {
                Schema::Bool(true)
            }
        }

        #[test]
        fn step_event_no_rust_type_without_info() {
            let mut generator = SchemaGenerator::default();
            let schema =
                StepEvent::<super::TestSpec>::json_schema(&mut generator);
            assert!(
                get_rust_type_ext(&schema).is_none(),
                "no x-rust-type when spec returns None"
            );
        }

        #[test]
        fn progress_event_no_rust_type_without_info() {
            let mut generator = SchemaGenerator::default();
            let schema =
                ProgressEvent::<super::TestSpec>::json_schema(&mut generator);
            assert!(
                get_rust_type_ext(&schema).is_none(),
                "no x-rust-type when spec returns None"
            );
        }

        #[test]
        fn event_report_no_rust_type_without_info() {
            let mut generator = SchemaGenerator::default();
            let schema =
                EventReport::<super::TestSpec>::json_schema(&mut generator);
            assert!(
                get_rust_type_ext(&schema).is_none(),
                "no x-rust-type when spec returns None"
            );
        }

        // -- Generic types with a spec whose rust_type_info points
        // to an external crate --

        /// A spec that returns `RustTypeInfo` pointing to a
        /// hypothetical external crate, simulating a user-defined
        /// spec in a downstream consumer.
        enum TestSpecWithInfo {}

        impl crate::spec::EngineSpec for TestSpecWithInfo {
            fn spec_name() -> String {
                "TestSpecWithInfo".to_owned()
            }

            type Component = serde_json::Value;
            type StepId = serde_json::Value;
            type StepMetadata = serde_json::Value;
            type ProgressMetadata = serde_json::Value;
            type CompletionMetadata = serde_json::Value;
            type SkippedMetadata = serde_json::Value;
            type Error = anyhow::Error;

            fn rust_type_info() -> Option<RustTypeInfo> {
                Some(RustTypeInfo {
                    crate_name: "my-external-crate",
                    version: "1.2.3",
                    path: "my_external_crate::MySpec",
                })
            }
        }

        impl JsonSchema for TestSpecWithInfo {
            fn schema_name() -> String {
                "TestSpecWithInfo".to_owned()
            }

            fn json_schema(_: &mut SchemaGenerator) -> Schema {
                Schema::Bool(true)
            }
        }

        /// Assert that the outer type in an x-rust-type extension
        /// points to `oxide-update-engine-types`, while the
        /// parameter points to the spec's crate.
        fn assert_external_spec_rust_type(
            x_rust_type: &serde_json::Value,
            expected_outer_path: &str,
            expected_param_crate: &str,
            expected_param_version: &str,
            expected_param_path: &str,
        ) {
            // Outer type: always oxide-update-engine-types.
            assert_rust_type_ext(
                x_rust_type,
                "oxide-update-engine-types",
                env!("CARGO_PKG_VERSION"),
                expected_outer_path,
            );

            // Parameter: the spec's crate.
            let params = x_rust_type
                .get("parameters")
                .and_then(|v| v.as_array())
                .expect("parameters array present");
            assert_eq!(params.len(), 1, "exactly one parameter");
            let param_xrt = params[0]
                .get("x-rust-type")
                .expect("parameter x-rust-type present");
            assert_rust_type_ext(
                param_xrt,
                expected_param_crate,
                expected_param_version,
                expected_param_path,
            );
        }

        #[test]
        fn step_event_external_spec_rust_type() {
            let mut generator = SchemaGenerator::default();
            let schema =
                StepEvent::<TestSpecWithInfo>::json_schema(&mut generator);
            let xrt = get_rust_type_ext(&schema).expect("x-rust-type present");
            assert_external_spec_rust_type(
                xrt,
                "oxide_update_engine_types::events::StepEvent",
                "my-external-crate",
                "1.2.3",
                "my_external_crate::MySpec",
            );
        }

        #[test]
        fn progress_event_external_spec_rust_type() {
            let mut generator = SchemaGenerator::default();
            let schema =
                ProgressEvent::<TestSpecWithInfo>::json_schema(&mut generator);
            let xrt = get_rust_type_ext(&schema).expect("x-rust-type present");
            assert_external_spec_rust_type(
                xrt,
                "oxide_update_engine_types::events\
                 ::ProgressEvent",
                "my-external-crate",
                "1.2.3",
                "my_external_crate::MySpec",
            );
        }

        #[test]
        fn event_report_external_spec_rust_type() {
            let mut generator = SchemaGenerator::default();
            let schema =
                EventReport::<TestSpecWithInfo>::json_schema(&mut generator);
            let xrt = get_rust_type_ext(&schema).expect("x-rust-type present");
            assert_external_spec_rust_type(
                xrt,
                "oxide_update_engine_types::events\
                 ::EventReport",
                "my-external-crate",
                "1.2.3",
                "my_external_crate::MySpec",
            );
        }
    }
}