aion-server 0.31.0

//! Turning one settled dispatch into the engine-facing reason string.
//!
//! Two of those reasons are TRANSPORT-domain classifications — a worker lost
//! before it reported, and a worker that refused the dispatch it had no slot
//! for. Both say the activity never ran, both ride the shared transport-loss
//! ledger, and both are worded apart so an operator is not sent looking for a
//! fault that is not there. The completion funnel that consumes them, and the
//! payload decode its success arm needs, live here with them.
//!
//! Split out of `bridge.rs`: this is one closed cluster with a single entry
//! point, and keeping it beside the dispatcher pushed that file past the
//! per-file length budget.

use aion_core::{ActivityErrorKind, ActivityId, ContentType, Payload, WorkflowId};

use crate::error::ServerError;
use crate::worker::dispatch::{ActivityCompletion, ActivityCompletionOutcome};

use super::PendingActivities;

impl PendingActivities {
    /// The transport-loss ledger this sink classifies worker deaths through.
    #[must_use]
    pub const fn transport_losses(&self) -> &crate::worker::transport_loss::TransportLossLedger {
        &self.transport_losses
    }

    /// Classify one worker loss for `(workflow_id, activity_id)` into the
    /// transport-domain reason the engine seam consumes.
    ///
    /// A ledger failure (poisoned lock) is reported as transport exhaustion
    /// rather than as a re-dispatchable loss: with no trustworthy budget the
    /// only safe answer is the one that terminates, because an unbounded
    /// re-dispatch is the failure mode the budget exists to prevent.
    pub(super) fn classify_worker_loss(
        &self,
        workflow_id: &WorkflowId,
        activity_id: &ActivityId,
        worker_id: crate::worker::registry::WorkerId,
    ) -> String {
        let detail = crate::worker::transport_loss::worker_lost_detail(worker_id);
        match self
            .transport_losses
            .record_loss(workflow_id, activity_id, &detail)
        {
            Ok(verdict) => {
                if verdict.exhausted {
                    tracing::error!(
                        operation = "activity_complete",
                        workflow_id = %workflow_id,
                        activity_id = %activity_id,
                        worker_id = ?worker_id,
                        error_type = "TransportExhausted",
                        losses = verdict.losses,
                        budget_ms = self.transport_losses.budget().as_millis(),
                        "activity abandoned: the transport kept losing its worker past the \
                         transport-loss budget"
                    );
                } else {
                    tracing::warn!(
                        operation = "activity_complete",
                        workflow_id = %workflow_id,
                        activity_id = %activity_id,
                        worker_id = ?worker_id,
                        error_type = "WorkerLost",
                        losses = verdict.losses,
                        budget_ms = self.transport_losses.budget().as_millis(),
                        "worker lost before reporting an activity result; the activity never ran \
                         and will be re-dispatched attempt-neutrally"
                    );
                }
                verdict.reason
            }
            Err(error) => {
                tracing::error!(
                    workflow_id = %workflow_id,
                    activity_id = %activity_id,
                    %error,
                    "transport-loss ledger is unreadable; abandoning the activity rather than \
                     re-dispatching it without a budget"
                );
                format!(
                    "{}{detail} (transport-loss budget unreadable: {error})",
                    crate::worker::transport_loss::TRANSPORT_EXHAUSTED_REASON_PREFIX
                )
            }
        }
    }

    /// Classify a worker's admission refusal into the TRANSPORT domain.
    ///
    /// Same domain as a worker loss and the same engine-facing consequence —
    /// the activity never ran, so it is re-dispatched attempt-neutrally and the
    /// action's authored retry budget is untouched — but a different fact, said
    /// differently: nothing died, and an operator told "worker lost" about a
    /// healthy worker would go looking for a fault that is not there.
    ///
    /// It rides the SAME transport-loss ledger, and that is deliberate rather
    /// than incidental. Attempt-neutral must not mean unbounded: with capacity
    /// on the wire the server does not select a full worker, so a refusal means
    /// the two counts have drifted, and a drift that keeps recurring is a real
    /// pathology that has to terminate rather than re-dispatch for ever. The
    /// ledger is the ceiling that already exists for exactly that shape, and
    /// giving this its own would be a second bound for one job.
    ///
    /// Logged at INFO, not WARN: a single refusal is a correction the system
    /// made for itself, and the ledger's own escalation is what says when it
    /// has stopped being one.
    fn classify_admission_refusal(
        &self,
        workflow_id: &WorkflowId,
        activity_id: &ActivityId,
        worker_id: crate::worker::registry::WorkerId,
        reason: &str,
    ) -> String {
        let detail = format!("worker {worker_id:?} refused the dispatch: {reason}");
        match self
            .transport_losses
            .record_loss(workflow_id, activity_id, &detail)
        {
            Ok(verdict) if verdict.exhausted => {
                tracing::error!(
                    operation = "activity_complete",
                    workflow_id = %workflow_id,
                    activity_id = %activity_id,
                    worker_id = ?worker_id,
                    error_type = "TransportExhausted",
                    refusals = verdict.losses,
                    budget_ms = self.transport_losses.budget().as_millis(),
                    reason,
                    "activity abandoned: workers kept refusing it past the transport-loss budget.                      The server tracks every worker's advertised concurrency and does not select                      a full one, so a refusal this persistent means the server's count and the                      worker's admission disagree"
                );
                verdict.reason
            }
            Ok(verdict) => {
                tracing::info!(
                    operation = "activity_complete",
                    workflow_id = %workflow_id,
                    activity_id = %activity_id,
                    worker_id = ?worker_id,
                    refusals = verdict.losses,
                    reason,
                    "worker refused a dispatch it had no slot for; re-parking it clock-free and                      re-selecting, with no attempt consumed and nothing recorded"
                );
                verdict.reason
            }
            Err(error) => {
                tracing::error!(
                    workflow_id = %workflow_id,
                    activity_id = %activity_id,
                    %error,
                    "transport-loss ledger is unreadable; abandoning the refused activity rather                      than re-dispatching it without a budget"
                );
                format!(
                    "{}{detail} (transport-loss budget unreadable: {error})",
                    crate::worker::transport_loss::TRANSPORT_EXHAUSTED_REASON_PREFIX
                )
            }
        }
    }

    pub(crate) fn complete_activity_after_accept(
        &self,
        completion: ActivityCompletion,
        after_accept: impl FnOnce() -> Result<(), ServerError>,
    ) -> Result<(), ServerError> {
        let result = match completion.outcome {
            ActivityCompletionOutcome::Succeeded(payload) => {
                payload_to_string(&payload).map_err(|reason| {
                    tracing::error!(
                        operation = "activity_complete",
                        workflow_id = %completion.workflow_id,
                        activity_id = %completion.activity_id,
                        error_type = "ActivityResultDecode",
                        %reason,
                        "activity completion failed"
                    );
                    ServerError::worker_dispatch("", "", format!("payload decode: {reason}"))
                })?
            }
            ActivityCompletionOutcome::Failed(error) => {
                let prefix = match error.kind {
                    ActivityErrorKind::Retryable => "retryable",
                    ActivityErrorKind::PolicyRefused => "policy_refused",
                    ActivityErrorKind::Terminal => "terminal",
                };
                tracing::error!(
                    operation = "activity_complete",
                    workflow_id = %completion.workflow_id,
                    activity_id = %completion.activity_id,
                    error_type = "ActivityFailed",
                    error_kind = prefix,
                    reason = %error.message,
                    "activity completion failed"
                );
                Err(format!("{prefix}:{}", error.message))
            }
            ActivityCompletionOutcome::WorkerLost { worker_id } => Err(self.classify_worker_loss(
                &completion.workflow_id,
                &completion.activity_id,
                worker_id,
            )),
            ActivityCompletionOutcome::Refused { worker_id, reason } => Err(self
                .classify_admission_refusal(
                    &completion.workflow_id,
                    &completion.activity_id,
                    worker_id,
                    &reason,
                )),
        };
        let accepted_settlement = || after_accept().map(|()| true);
        self.complete_fenced_after_accept(
            &completion.workflow_id,
            &completion.activity_id,
            completion.run_id.as_ref(),
            &completion.completion_token,
            result,
            accepted_settlement,
        )?;
        Ok(())
    }
}

fn payload_to_string(payload: &Payload) -> Result<Result<String, String>, String> {
    match payload.content_type() {
        ContentType::Json => String::from_utf8(payload.bytes().to_vec())
            .map(Ok)
            .map_err(|_| "activity result payload is not valid UTF-8".to_owned()),
    }
}