oxide_batch_repository/

operator.rs

//! The durable operator request, audit record, and guard vocabulary.
//!
//! Every mutating action carries a bounded envelope and commits its append-only
//! audit row in the same transaction as its effect. The values below are what a
//! metadata adapter persists and replays; the service that applies them lives
//! above this crate.

use std::fmt;
use std::time::SystemTime;

use oxide_batch_core::{
    BatchStatus, DefinitionIdentity, ExecutionVersion, FailureSummary, JobExecutionId,
    JobInstanceId, JobInstanceKey, LifecycleError, OperatorRequestId,
};

use crate::{
    ActorRef, AuthorizationClass, OperationId, OperatorAction, ReasonCode, RecoveryDisposition,
    RecoveryProposal, RecoveryRequest, RecoveryRequestError, RepositoryError, RequestArguments,
    RequestDigest, hex_digest, request_digest,
};

/// One validated mutating operator request.
///
/// The request digest covers the action, target identity, expected version,
/// and bounded arguments. Replaying an operation identifier with a different
/// digest is a conflict rather than a repeat.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct OperatorRequest {
    action: OperatorAction,
    operation_id: OperationId,
    actor: ActorRef,
    reason: Option<ReasonCode>,
    target: OperatorTarget,
    expected_version: Option<ExecutionVersion>,
    arguments: RequestArguments,
    digest: RequestDigest,
}

/// The disposition of one recovery decision together with the evidence that
/// disposition requires.
///
/// Pairing the two makes a `MarkFailed` decision without its stated failure
/// unrepresentable rather than a deferred validation error, and keeps an
/// `Abandon` decision from carrying a failure that its durable outcome ignores
/// but its request digest would still cover.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub enum RecoveryDirective {
    /// Make the observed attempt restart-eligible under a stated failure.
    MarkFailed(FailureSummary),
    /// Make the logical instance permanently non-restartable.
    Abandon,
}

impl RecoveryDirective {
    /// Returns the durable disposition this directive requests.
    #[must_use]
    pub const fn disposition(self) -> RecoveryDisposition {
        match self {
            Self::MarkFailed(_) => RecoveryDisposition::MarkFailed,
            Self::Abandon => RecoveryDisposition::Abandon,
        }
    }

    /// Returns the stated failure of a `MarkFailed` directive.
    #[must_use]
    pub const fn failure(self) -> Option<FailureSummary> {
        match self {
            Self::MarkFailed(failure) => Some(failure),
            Self::Abandon => None,
        }
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
enum OperatorTarget {
    InstanceKey(Box<JobInstanceKey>),
    Instance(JobInstanceId),
    Execution(JobExecutionId),
}

impl OperatorTarget {
    fn identity(&self) -> String {
        match self {
            Self::InstanceKey(key) => {
                format!("instance-key:{}", hex_digest(&key.digest()))
            }
            Self::Instance(id) => format!("instance:{id}"),
            Self::Execution(id) => format!("execution:{id}"),
        }
    }
}

impl OperatorRequest {
    /// Requests one launch of the instance selected by an identifying key.
    #[must_use]
    pub fn launch(
        operation_id: OperationId,
        actor: ActorRef,
        key: JobInstanceKey,
        definition: DefinitionIdentity,
    ) -> Self {
        Self::build(
            OperatorAction::Launch,
            operation_id,
            actor,
            None,
            OperatorTarget::InstanceKey(Box::new(key)),
            None,
            RequestArguments::Definition(Box::new(definition)),
        )
    }

    /// Requests one restart attempt for an existing logical instance.
    #[must_use]
    pub fn restart(
        operation_id: OperationId,
        actor: ActorRef,
        job_instance_id: JobInstanceId,
        definition: DefinitionIdentity,
    ) -> Self {
        Self::build(
            OperatorAction::Restart,
            operation_id,
            actor,
            None,
            OperatorTarget::Instance(job_instance_id),
            None,
            RequestArguments::Definition(Box::new(definition)),
        )
    }

    /// Requests one durable cooperative stop.
    #[must_use]
    pub fn stop(
        operation_id: OperationId,
        actor: ActorRef,
        job_execution_id: JobExecutionId,
        expected_version: ExecutionVersion,
    ) -> Self {
        Self::build(
            OperatorAction::Stop,
            operation_id,
            actor,
            None,
            OperatorTarget::Execution(job_execution_id),
            Some(expected_version),
            RequestArguments::None,
        )
    }

    /// Requests that one finished or recovered execution become `ABANDONED`.
    #[must_use]
    pub fn abandon(
        operation_id: OperationId,
        actor: ActorRef,
        reason: ReasonCode,
        job_execution_id: JobExecutionId,
        expected_version: ExecutionVersion,
    ) -> Self {
        Self::build(
            OperatorAction::Abandon,
            operation_id,
            actor,
            Some(reason),
            OperatorTarget::Execution(job_execution_id),
            Some(expected_version),
            RequestArguments::None,
        )
    }

    /// Requests one evidence-bound recovery decision.
    #[must_use]
    pub fn recover(
        operation_id: OperationId,
        actor: ActorRef,
        reason: ReasonCode,
        directive: RecoveryDirective,
        proposal: &RecoveryProposal,
    ) -> Self {
        let job_execution_id = proposal.evidence().execution_id();
        let expected_version = proposal.observed_version();
        let evidence_digest = *proposal.digest();
        Self::build(
            OperatorAction::Recover,
            operation_id,
            actor,
            Some(reason),
            OperatorTarget::Execution(job_execution_id),
            Some(expected_version),
            RequestArguments::Recovery {
                directive,
                evidence_digest,
                unknown_commit: proposal.evidence().unknown_commit(),
            },
        )
    }

    fn build(
        action: OperatorAction,
        operation_id: OperationId,
        actor: ActorRef,
        reason: Option<ReasonCode>,
        target: OperatorTarget,
        expected_version: Option<ExecutionVersion>,
        arguments: RequestArguments,
    ) -> Self {
        let digest = request_digest(
            action,
            &target.identity(),
            expected_version,
            reason.as_ref(),
            &arguments,
        );
        Self {
            action,
            operation_id,
            actor,
            reason,
            target,
            expected_version,
            arguments,
            digest,
        }
    }

    /// Returns the requested action.
    #[must_use]
    pub const fn action(&self) -> OperatorAction {
        self.action
    }

    /// Returns the class a deployment authorizes before this call.
    #[must_use]
    pub const fn authorization_class(&self) -> AuthorizationClass {
        self.action.authorization_class()
    }

    /// Borrows the caller-supplied idempotency key.
    #[must_use]
    pub const fn operation_id(&self) -> &OperationId {
        &self.operation_id
    }

    /// Borrows the opaque actor reference.
    #[must_use]
    pub const fn actor(&self) -> &ActorRef {
        &self.actor
    }

    /// Borrows the closed-set reason code, when the action requires one.
    #[must_use]
    pub const fn reason(&self) -> Option<&ReasonCode> {
        self.reason.as_ref()
    }

    /// Returns the observed optimistic version for a lifecycle mutation.
    #[must_use]
    pub const fn expected_version(&self) -> Option<ExecutionVersion> {
        self.expected_version
    }

    /// Returns the framework-computed canonical request digest.
    #[must_use]
    pub const fn digest(&self) -> &RequestDigest {
        &self.digest
    }

    /// Returns the targeted execution, when the action names one.
    #[must_use]
    pub const fn job_execution_id(&self) -> Option<JobExecutionId> {
        match self.target {
            OperatorTarget::Execution(id) => Some(id),
            _ => None,
        }
    }

    /// Returns the targeted logical instance, when the action names one.
    #[must_use]
    pub const fn job_instance_id(&self) -> Option<JobInstanceId> {
        match self.target {
            OperatorTarget::Instance(id) => Some(id),
            _ => None,
        }
    }

    /// Borrows the identifying key, when the action selects by key.
    ///
    /// Only [`OperatorAction::Launch`] names one; every other action names an
    /// instance or an execution.
    #[must_use]
    pub fn job_instance_key(&self) -> Option<&JobInstanceKey> {
        match &self.target {
            OperatorTarget::InstanceKey(key) => Some(key),
            _ => None,
        }
    }

    /// Borrows the definition identity this request launches with, if any.
    #[doc(hidden)]
    #[must_use]
    pub fn definition(&self) -> Option<&DefinitionIdentity> {
        match &self.arguments {
            RequestArguments::Definition(definition) => Some(definition),
            _ => None,
        }
    }

    /// Builds the durable recovery request this operator request carries.
    #[doc(hidden)]
    #[must_use]
    pub fn recovery_request(&self) -> Option<Result<RecoveryRequest, RecoveryRequestError>> {
        let RequestArguments::Recovery {
            directive,
            evidence_digest,
            ..
        } = &self.arguments
        else {
            return None;
        };
        let expected_version = self.expected_version?;
        let reason = self.reason.as_ref()?;
        Some(match directive {
            RecoveryDirective::Abandon => RecoveryRequest::abandon(
                expected_version,
                reason.as_str(),
                self.actor.as_str(),
                *evidence_digest,
            ),
            RecoveryDirective::MarkFailed(failure) => RecoveryRequest::mark_failed(
                expected_version,
                reason.as_str(),
                self.actor.as_str(),
                *evidence_digest,
                failure.category(),
                failure.failure_id(),
            ),
        })
    }

    /// Borrows the recovery guard this request must satisfy, when it has one.
    #[doc(hidden)]
    #[must_use]
    pub fn recovery_guard(&self) -> Option<(RecoveryDirective, bool)> {
        match &self.arguments {
            RequestArguments::Recovery {
                directive,
                unknown_commit,
                ..
            } => Some((*directive, *unknown_commit)),
            _ => None,
        }
    }
}

/// The durable class of one recorded operator request.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
#[non_exhaustive]
pub enum OperatorOutcomeClass {
    /// The request was guarded, applied, and audited.
    Applied,
    /// A durable record for this operation identifier already existed.
    Replayed,
    /// A guard rejected the request; the audit row records the class.
    Rejected,
}

impl OperatorOutcomeClass {
    /// Returns the stable durable code for this class.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Applied => "APPLIED",
            Self::Replayed => "REPLAYED",
            Self::Rejected => "REJECTED",
        }
    }
}

/// The typed reason one guard rejected an operator action.
///
/// A rejection is durable and audited. It carries no user error text, SQL, or
/// credential.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub enum OperatorRejection {
    /// The supplied expected version lost its compare-and-swap.
    OptimisticConflict {
        /// The version observed under lock.
        current: ExecutionVersion,
    },
    /// The action is not legal from the observed status.
    InvalidState {
        /// The status observed under lock.
        status: BatchStatus,
    },
    /// The logical instance already completed.
    InstanceCompleted,
    /// The logical instance is permanently abandoned.
    InstanceAbandoned,
    /// Another attempt is active or requires explicit recovery.
    ExecutionAlreadyActive {
        /// The attempt preventing the action.
        execution_id: JobExecutionId,
        /// Its status observed under lock.
        status: BatchStatus,
    },
    /// The proposed definition cannot interpret the committed checkpoint.
    IncompatibleDefinition,
    /// A restart was requested for an instance with no prior attempt.
    RestartWithoutPriorAttempt,
    /// The instance-wide start limit for a logical step is exhausted.
    StartLimitExceeded,
    /// Abandoning an ambiguous execution requires an applied recovery decision.
    UnresolvedRecoveryRequired,
    /// The targeted execution does not exist.
    ExecutionNotFound,
    /// The targeted logical instance does not exist.
    InstanceNotFound,
    /// This build cannot apply the requested action.
    ///
    /// [`OperatorAction`] is `#[non_exhaustive]`, so a caller compiled against
    /// a newer definition of it can name an action this build has no effect
    /// for. Rejecting is the conservative arm: the request is audited and
    /// nothing is applied.
    UnsupportedAction,
}

impl OperatorRejection {
    /// Returns the stable durable code for this rejection.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::OptimisticConflict { .. } => "OPTIMISTIC_CONFLICT",
            Self::InvalidState { .. } => "INVALID_STATE",
            Self::InstanceCompleted => "INSTANCE_COMPLETED",
            Self::InstanceAbandoned => "INSTANCE_ABANDONED",
            Self::ExecutionAlreadyActive { .. } => "EXECUTION_ALREADY_ACTIVE",
            Self::IncompatibleDefinition => "INCOMPATIBLE_DEFINITION",
            Self::RestartWithoutPriorAttempt => "RESTART_WITHOUT_PRIOR_ATTEMPT",
            Self::StartLimitExceeded => "START_LIMIT_EXCEEDED",
            Self::UnresolvedRecoveryRequired => "UNRESOLVED_RECOVERY_REQUIRED",
            Self::ExecutionNotFound => "EXECUTION_NOT_FOUND",
            Self::InstanceNotFound => "INSTANCE_NOT_FOUND",
            Self::UnsupportedAction => "UNSUPPORTED_ACTION",
        }
    }

    /// Classifies one repository failure as a guard rejection, when it is one.
    #[doc(hidden)]
    #[must_use]
    pub fn from_repository(error: &RepositoryError) -> Option<Self> {
        match error {
            RepositoryError::CompletedInstance { .. } => Some(Self::InstanceCompleted),
            RepositoryError::AbandonedInstance { .. } => Some(Self::InstanceAbandoned),
            RepositoryError::ExecutionAlreadyActive {
                execution_id,
                status,
                ..
            } => Some(Self::ExecutionAlreadyActive {
                execution_id: *execution_id,
                status: *status,
            }),
            RepositoryError::IncompatibleDefinition { .. }
            | RepositoryError::RestartStateNotFound { .. }
            | RepositoryError::InvalidDefinitionUpgrade { .. } => {
                Some(Self::IncompatibleDefinition)
            }
            RepositoryError::StartLimitExceeded { .. } => Some(Self::StartLimitExceeded),
            RepositoryError::JobExecutionNotFound { .. } => Some(Self::ExecutionNotFound),
            RepositoryError::JobInstanceNotFound { .. } => Some(Self::InstanceNotFound),

            RepositoryError::Lifecycle(LifecycleError::StaleVersion { actual, .. }) => {
                Some(Self::OptimisticConflict { current: *actual })
            }
            RepositoryError::Lifecycle(
                LifecycleError::IllegalTransition { from, .. }
                | LifecycleError::RestartRequiresNewAttempt { from },
            ) => Some(Self::InvalidState { status: *from }),
            RepositoryError::RecoveryNotAllowed { status, .. }
            | RepositoryError::Lifecycle(LifecycleError::NotRestartable { status }) => {
                Some(Self::InvalidState { status: *status })
            }
            _ => None,
        }
    }
}

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

/// One append-only operator audit and idempotency record.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct OperatorRecord {
    id: OperatorRequestId,
    action: OperatorAction,
    operation_id: OperationId,
    actor: ActorRef,
    reason: Option<ReasonCode>,
    digest: RequestDigest,
    job_instance_id: Option<JobInstanceId>,
    job_execution_id: Option<JobExecutionId>,
    observed_version: Option<ExecutionVersion>,
    prior_status: Option<BatchStatus>,
    result_status: Option<BatchStatus>,
    outcome: OperatorOutcomeClass,
    rejection: Option<OperatorRejection>,
    requested_at: SystemTime,
}

impl OperatorRecord {
    /// Rebuilds a record read from a durable adapter.
    #[must_use]
    pub fn from_parts(id: OperatorRequestId, draft: OperatorRecordDraft) -> Self {
        Self {
            id,
            action: draft.action,
            operation_id: draft.operation_id,
            actor: draft.actor,
            reason: draft.reason,
            digest: draft.digest,
            job_instance_id: draft.job_instance_id,
            job_execution_id: draft.job_execution_id,
            observed_version: draft.observed_version,
            prior_status: draft.prior_status,
            result_status: draft.result_status,
            outcome: draft.outcome,
            rejection: draft.rejection,
            requested_at: draft.requested_at,
        }
    }

    /// Returns the opaque record identifier.
    #[must_use]
    pub const fn id(&self) -> OperatorRequestId {
        self.id
    }

    /// Returns the audited action.
    #[must_use]
    pub const fn action(&self) -> OperatorAction {
        self.action
    }

    /// Borrows the idempotency key.
    #[must_use]
    pub const fn operation_id(&self) -> &OperationId {
        &self.operation_id
    }

    /// Borrows the opaque actor reference.
    #[must_use]
    pub const fn actor(&self) -> &ActorRef {
        &self.actor
    }

    /// Borrows the closed-set reason code, when the action required one.
    #[must_use]
    pub const fn reason(&self) -> Option<&ReasonCode> {
        self.reason.as_ref()
    }

    /// Returns the recorded canonical request digest.
    #[must_use]
    pub const fn digest(&self) -> &RequestDigest {
        &self.digest
    }

    /// Returns the audited logical instance, when the action named one.
    #[must_use]
    pub const fn job_instance_id(&self) -> Option<JobInstanceId> {
        self.job_instance_id
    }

    /// Returns the audited execution, when the action produced or named one.
    #[must_use]
    pub const fn job_execution_id(&self) -> Option<JobExecutionId> {
        self.job_execution_id
    }

    /// Returns the version observed under lock.
    #[must_use]
    pub const fn observed_version(&self) -> Option<ExecutionVersion> {
        self.observed_version
    }

    /// Returns the status observed before the effect.
    #[must_use]
    pub const fn prior_status(&self) -> Option<BatchStatus> {
        self.prior_status
    }

    /// Returns the status the effect produced.
    #[must_use]
    pub const fn result_status(&self) -> Option<BatchStatus> {
        self.result_status
    }

    /// Returns the recorded outcome class.
    #[must_use]
    pub const fn outcome(&self) -> OperatorOutcomeClass {
        self.outcome
    }

    /// Returns the recorded rejection class, when the action was rejected.
    #[must_use]
    pub const fn rejection(&self) -> Option<OperatorRejection> {
        self.rejection
    }

    /// Returns the facade-clock instant recorded with the request.
    #[must_use]
    pub const fn requested_at(&self) -> SystemTime {
        self.requested_at
    }
}

/// The bounded audit row an adapter appends.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct OperatorRecordDraft {
    action: OperatorAction,
    operation_id: OperationId,
    actor: ActorRef,
    reason: Option<ReasonCode>,
    digest: RequestDigest,
    job_instance_id: Option<JobInstanceId>,
    job_execution_id: Option<JobExecutionId>,
    observed_version: Option<ExecutionVersion>,
    prior_status: Option<BatchStatus>,
    result_status: Option<BatchStatus>,
    outcome: OperatorOutcomeClass,
    rejection: Option<OperatorRejection>,
    requested_at: SystemTime,
}

impl OperatorRecordDraft {
    /// Drafts the audit row for one applied request.
    ///
    /// The audited action, actor, reason, and digest are taken from the
    /// request, so the row cannot disagree with the request it audits. The
    /// four effect values are what the transaction observed and produced.
    #[must_use]
    pub fn applied(
        request: &OperatorRequest,
        job_instance_id: Option<JobInstanceId>,
        job_execution_id: Option<JobExecutionId>,
        prior_status: Option<BatchStatus>,
        result_status: Option<BatchStatus>,
        requested_at: SystemTime,
    ) -> Self {
        Self {
            action: request.action(),
            operation_id: request.operation_id().clone(),
            actor: request.actor().clone(),
            reason: request.reason().cloned(),
            digest: *request.digest(),
            job_instance_id,
            job_execution_id,
            observed_version: request.expected_version(),
            prior_status,
            result_status,
            outcome: OperatorOutcomeClass::Applied,
            rejection: None,
            requested_at,
        }
    }

    /// Drafts the audit row for one rejected request.
    ///
    /// A rejection applies no effect, so the row carries only the target the
    /// request already named and no observed or produced status.
    #[must_use]
    pub fn rejected(
        request: &OperatorRequest,
        rejection: OperatorRejection,
        requested_at: SystemTime,
    ) -> Self {
        Self {
            action: request.action(),
            operation_id: request.operation_id().clone(),
            actor: request.actor().clone(),
            reason: request.reason().cloned(),
            digest: *request.digest(),
            job_instance_id: request.job_instance_id(),
            job_execution_id: request.job_execution_id(),
            observed_version: request.expected_version(),
            prior_status: None,
            result_status: None,
            outcome: OperatorOutcomeClass::Rejected,
            rejection: Some(rejection),
            requested_at,
        }
    }

    /// Rebuilds a draft from one durable audit row.
    ///
    /// Durable adapters use this to return a recorded outcome without
    /// re-deriving it from a request that may no longer exist.
    #[must_use]
    #[allow(clippy::too_many_arguments)]
    pub const fn from_durable(
        action: OperatorAction,
        operation_id: OperationId,
        actor: ActorRef,
        reason: Option<ReasonCode>,
        digest: RequestDigest,
        job_instance_id: Option<JobInstanceId>,
        job_execution_id: Option<JobExecutionId>,
        observed_version: Option<ExecutionVersion>,
        prior_status: Option<BatchStatus>,
        result_status: Option<BatchStatus>,
        outcome: OperatorOutcomeClass,
        rejection: Option<OperatorRejection>,
        requested_at: SystemTime,
    ) -> Self {
        Self {
            action,
            operation_id,
            actor,
            reason,
            digest,
            job_instance_id,
            job_execution_id,
            observed_version,
            prior_status,
            result_status,
            outcome,
            rejection,
            requested_at,
        }
    }

    /// Returns the audited action.
    #[must_use]
    pub const fn action(&self) -> OperatorAction {
        self.action
    }

    /// Borrows the idempotency key.
    #[must_use]
    pub const fn operation_id(&self) -> &OperationId {
        &self.operation_id
    }

    /// Borrows the opaque actor reference.
    #[must_use]
    pub const fn actor(&self) -> &ActorRef {
        &self.actor
    }

    /// Borrows the closed-set reason code, when present.
    #[must_use]
    pub const fn reason(&self) -> Option<&ReasonCode> {
        self.reason.as_ref()
    }

    /// Returns the canonical request digest.
    #[must_use]
    pub const fn digest(&self) -> &RequestDigest {
        &self.digest
    }

    /// Returns the audited logical instance, when known.
    #[must_use]
    pub const fn job_instance_id(&self) -> Option<JobInstanceId> {
        self.job_instance_id
    }

    /// Returns the audited execution, when known.
    #[must_use]
    pub const fn job_execution_id(&self) -> Option<JobExecutionId> {
        self.job_execution_id
    }

    /// Returns the version observed under lock.
    #[must_use]
    pub const fn observed_version(&self) -> Option<ExecutionVersion> {
        self.observed_version
    }

    /// Returns the status observed before the effect.
    #[must_use]
    pub const fn prior_status(&self) -> Option<BatchStatus> {
        self.prior_status
    }

    /// Returns the status the effect produced.
    #[must_use]
    pub const fn result_status(&self) -> Option<BatchStatus> {
        self.result_status
    }

    /// Returns the recorded outcome class.
    #[must_use]
    pub const fn outcome(&self) -> OperatorOutcomeClass {
        self.outcome
    }

    /// Returns the recorded rejection class, when the action was rejected.
    #[must_use]
    pub const fn rejection(&self) -> Option<OperatorRejection> {
        self.rejection
    }

    /// Returns the facade-clock instant recorded with the request.
    #[must_use]
    pub const fn requested_at(&self) -> SystemTime {
        self.requested_at
    }
}