aion-server 0.29.0

//! Describe-workflow handler.

use aion_core::{Event, ReadProvenance, WorkflowSummary, current_lease_terminal};
use aion_proto::{
    ProtoDescribeWorkflowRequest, ProtoDescribeWorkflowResponse, WireError,
    convert::{encode_event, encode_workflow_summary},
};

use super::error::workflow_not_found_error;
use super::payload::{encode_history, required_workflow_id};
use super::runs::resolve_run_id;
use crate::{CallerIdentity, NamespaceGuard, NamespaceOperation, ServerError, WorkflowTarget};

/// The describe read's full result: the wire response every transport returns,
/// plus the two inputs a transport may need to project further.
///
/// The scoped `namespace` and the decoded `history` are handed back rather than
/// dropped because the HTTP surface projects an unserved-activity verdict over
/// them ([`crate::worker::ActivityReachability`]). Recomputing either at the
/// transport would mean a second namespace scope and a second `read_history` for
/// one request, and — worse — a second answer that could disagree with the one
/// the summary was built from.
#[derive(Debug)]
pub struct DescribeOutcome {
    /// Wire response: the encoded summary and (optionally) the encoded history.
    pub response: ProtoDescribeWorkflowResponse,
    /// The namespace the caller was scoped to.
    pub namespace: String,
    /// The workflow the read resolved to.
    pub workflow_id: aion_core::WorkflowId,
    /// The run's decoded history, read in full regardless of `include_history`.
    pub history: Vec<Event>,
}

/// Handles a decoded describe-workflow request.
///
/// # Errors
///
/// Returns a stable [`WireError`] when IDs are missing or malformed, namespace scoping fails, store
/// history reading fails, the workflow has no summary, or response envelopes cannot be encoded.
///
/// `provenance` is what the serving install says about ITSELF (ADR-016) and
/// is stamped on the response as given: it is a parameter rather than
/// something this handler reads, because the count lives with the worker
/// dispatcher, which a namespace-scoped read cannot see — and a boundary that
/// forgot to supply it would emit `0`, which reads as "never recorded here".
/// Making it an argument makes forgetting a compile error.
pub async fn describe(
    guard: &NamespaceGuard,
    caller: &CallerIdentity,
    request: ProtoDescribeWorkflowRequest,
    provenance: ReadProvenance,
) -> Result<DescribeOutcome, WireError> {
    let workflow_id = required_workflow_id(request.workflow_id.clone())?;
    let target = WorkflowTarget::workflow(&workflow_id);
    let scoped = guard
        .scope(caller, &NamespaceOperation::describe(&request, target))
        .await
        .map_err(|error| error.to_wire_error())?;
    let engine = scoped.engine().map_err(|error| error.to_wire_error())?;
    let run_id = resolve_run_id(engine.as_ref(), &workflow_id, request.run_id.clone()).await?;

    let history = engine
        .store()
        .read_history(&workflow_id)
        .await
        .map_err(|error| ServerError::from(error).to_wire_error())?;
    let summary = WorkflowSummary::from_history(&history)
        .ok_or_else(|| workflow_not_found_error(&workflow_id))?;
    let namespace = scoped.namespace().to_owned();
    let summary = encode_workflow_summary(namespace.clone(), None, &summary)?;
    let encoded_history = encode_history(request.include_history, &namespace, &history)?;
    let history_head_seq = history.last().map_or(0, Event::seq);
    let terminal_event = current_lease_terminal(&history)
        .map(|event| encode_event(namespace.clone(), None, event))
        .transpose()?;

    Ok(DescribeOutcome {
        response: ProtoDescribeWorkflowResponse {
            summary: Some(summary),
            history: encoded_history,
            run_id: Some(run_id.into()),
            history_head_seq,
            terminal_event,
            provenance: Some(provenance.into()),
            // Counted over the FULL history just read — the one place that
            // holds it — so every transport states the same fact.
            lease_recording: Some(aion_core::lease_recording(&history).into()),
        },
        namespace,
        workflow_id,
        history,
    })
}

#[cfg(test)]
mod tests {
    use aion_core::Event;
    use aion_proto::{
        WireError,
        convert::{decode_event, decode_workflow_summary},
    };

    use super::super::test_support::{
        NAMESPACE, append_started, assert_workflow_not_found, context, describe_request, run_id,
        workflow_id,
    };
    use super::describe;

    #[tokio::test]
    async fn describe_handler_scopes_then_reads_summary_and_optional_history()
    -> Result<(), Box<dyn std::error::Error>> {
        let context = context().await?;
        context.ownership.record(workflow_id(), NAMESPACE)?;
        append_started(context.store.as_ref()).await?;

        let outcome = describe(
            &context.guard,
            &context.caller,
            describe_request(true, None),
            aion_core::ReadProvenance::new(3),
        )
        .await?;
        // The decoded history and the scoped namespace come back for the
        // transport's own projection, and they are the SAME read the summary was
        // built from — not a second one that could disagree.
        assert_eq!(outcome.namespace, NAMESPACE);
        assert_eq!(outcome.history.len(), 1);
        let response = outcome.response;
        // ADR-016 / WA-010 R4: the install's own count is stamped as supplied.
        assert_eq!(
            response.provenance,
            Some(aion_proto::ProtoReadProvenance {
                lease_record_failures_total: 3
            })
        );
        // WA-010 R4: the whole-history lease counts are stated by the server.
        assert_eq!(
            response.lease_recording,
            Some(aion_proto::ProtoLeaseRecording::default()),
            "a started-only history has dispatched nothing and leased nothing"
        );

        let summary = response
            .summary
            .as_ref()
            .map(decode_workflow_summary)
            .transpose()?
            .ok_or_else(|| WireError::backend("summary missing"))?;
        assert_eq!(summary.workflow_id, workflow_id());
        assert_eq!(
            summary.package_version,
            Some(aion_core::PackageVersion::new("a".repeat(64))),
            "the summary names the package the run started under, from its own WorkflowStarted"
        );
        assert_eq!(response.history.len(), 1);
        assert!(matches!(
            decode_event(&response.history[0])?,
            Event::WorkflowStarted { .. }
        ));
        Ok(())
    }

    #[tokio::test]
    async fn describe_handler_maps_omitted_run_missing_workflow_to_not_found()
    -> Result<(), Box<dyn std::error::Error>> {
        let context = context().await?;
        context.ownership.record(workflow_id(), NAMESPACE)?;

        let error = describe(
            &context.guard,
            &context.caller,
            describe_request(false, None),
            aion_core::ReadProvenance::default(),
        )
        .await;

        assert_workflow_not_found(error)?;
        Ok(())
    }

    #[tokio::test]
    async fn describe_handler_maps_empty_history_to_not_found()
    -> Result<(), Box<dyn std::error::Error>> {
        let context = context().await?;
        context.ownership.record(workflow_id(), NAMESPACE)?;

        let error = describe(
            &context.guard,
            &context.caller,
            describe_request(false, Some(run_id())),
            aion_core::ReadProvenance::default(),
        )
        .await;

        assert_workflow_not_found(error)?;
        Ok(())
    }
}