aion-rs 0.10.0

//! Durable fan-out dispatch: the additive atomic write of the
//! `N×(ActivityScheduled + ActivityStarted)` event batch together with the matching `N` outbox
//! rows, in one store transaction.
//!
//! This is the Phase 1 Recorder capability behind the interim durable outbox. It is **additive**:
//! the live dispatch path (`nif_collect.rs::dispatch_unscheduled`) still records its events and
//! spawns completion tasks exactly as before. A later flag-gated cutover will route fresh fan-out
//! batches through [`Recorder::record_fan_out_dispatch`] instead.

use aion_core::{
    ActivityError, ActivityId, Event, EventEnvelope, Payload, RunId, status::run_segment,
};
use aion_store::OutboxRow;
use chrono::{DateTime, Utc};

use super::Recorder;
use crate::durability::DurabilityError;

/// Terminal outcome of a single fan-out activity, carrying exactly the fields the matching terminal
/// event records.
///
/// [`FanOutOutcome::Completed`] maps to [`Event::ActivityCompleted`] (its `result` payload and
/// one-based `attempt`); [`FanOutOutcome::Failed`] maps to [`Event::ActivityFailed`] (its classified
/// `error` and one-based `attempt`). The shapes mirror the events the live completion path records
/// today through
/// [`Recorder::record_activity_completed`] / [`Recorder::record_activity_failed`].
#[derive(Clone, Debug)]
pub enum FanOutOutcome {
    /// The activity succeeded with this result payload on the given one-based attempt.
    Completed {
        /// Opaque activity result payload.
        result: Payload,
        /// One-based activity attempt number that produced this completion (NOI-0).
        attempt: u32,
    },
    /// The activity attempt failed with this classified error on the given one-based attempt.
    Failed {
        /// Classified activity failure.
        error: ActivityError,
        /// One-based activity attempt number that produced this failure.
        attempt: u32,
    },
}

/// Outcome of [`Recorder::record_fan_out_completion`]: whether the terminal was newly recorded or
/// dropped as a duplicate of an already-resolved ordinal.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum FanOutCompletionResult {
    /// The terminal event was appended; the head advanced by exactly one.
    Recorded,
    /// The ordinal already had a terminal in history; nothing was appended and the head is unchanged.
    Dropped,
}

/// One member of a durable fan-out dispatch staged through [`Recorder::record_fan_out_dispatch`].
///
/// Each item carries the pinned `ordinal` within the workflow's contiguous fan-out range (the same
/// ordinal `nif_collect.rs` allocates), and the activity type and input the worker must execute. The
/// recorder derives the per-item `ActivityScheduled`/`ActivityStarted` events and the matching
/// [`OutboxRow`] (with `dispatch_key = "{workflow_id}:{ordinal}"`) from this.
#[derive(Clone, Debug)]
pub struct FanOutItem {
    /// Pinned ordinal of this activity within the workflow's fan-out range.
    pub ordinal: u64,
    /// Workflow's durable isolation namespace — the routing correctness boundary the staged outbox
    /// row records so the dispatcher routes within the workflow's real namespace (NSTQ-2).
    pub namespace: String,
    /// Pool/flavour selector for this dispatch. No SDK-level task-queue selection exists yet
    /// (NSTQ-4), so this is the named `"default"` task queue today (NSTQ-2).
    pub task_queue: String,
    /// OPTIONAL node affinity for this dispatch. No SDK-level node selection exists yet (NODE-4), so
    /// this is `None` today (NODE-2) — `None` = genuine current "no affinity", not a shim.
    pub node: Option<String>,
    /// Activity type the worker must execute.
    pub activity_type: String,
    /// Opaque activity input payload.
    pub input: Payload,
    /// One-based delivery attempt this fan-out dispatch belongs to (NOI-0). Recorded onto the
    /// derived `ActivityStarted` so it shares one `(workflow, activity, attempt)` identity with the
    /// terminal recorded through [`Recorder::record_fan_out_completion`]. No retry executor exists
    /// yet, so this is `FIRST_DELIVERY_ATTEMPT` (`1`) today — the genuine current value, not a shim.
    pub attempt: u32,
}

impl Recorder {
    /// Atomically records a durable fan-out dispatch: the `N×(ActivityScheduled + ActivityStarted)`
    /// event batch AND the matching `N` outbox rows in one store transaction.
    ///
    /// The event batch is the same shape the live dispatch path
    /// (`nif_collect.rs::dispatch_unscheduled`) records today — for each item, an
    /// [`Event::ActivityScheduled`] immediately followed by an [`Event::ActivityStarted`], both
    /// keyed by `ActivityId::from_sequence_position(ordinal)`, in `items` order. Each outbox row is a
    /// fresh [`OutboxRow::pending`] with the canonical `dispatch_key = "{workflow_id}:{ordinal}"`
    /// idempotency guard.
    ///
    /// On success the sequence head advances by exactly `2 * items.len()`. The store applies the
    /// events and outbox rows in a single transaction, so a failure (including
    /// [`StoreError::SequenceConflict`](aion_store::StoreError::SequenceConflict)) leaves the head
    /// unadvanced AND writes neither events nor outbox rows — matching the single-writer discipline
    /// where a conflict is a hard error, never a retry. An empty `items` slice is a no-op.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the event store rejects the atomic append (a sequence conflict
    /// surfaces without advancing the head), if an envelope sequence would overflow `u64`, or if the
    /// sequence tracker cannot advance after a successful append.
    pub async fn record_fan_out_dispatch(
        &mut self,
        recorded_at: DateTime<Utc>,
        items: &[FanOutItem],
    ) -> Result<(), DurabilityError> {
        if items.is_empty() {
            return Ok(());
        }

        let mut events = Vec::with_capacity(items.len() * 2);
        let mut outbox_rows = Vec::with_capacity(items.len());
        let mut previous: Option<EventEnvelope> = None;
        for item in items {
            let scheduled_envelope = match &previous {
                Some(previous) => self.envelope_after(previous, recorded_at)?,
                None => self.next_envelope(recorded_at)?,
            };
            let started_envelope = self.envelope_after(&scheduled_envelope, recorded_at)?;
            previous = Some(started_envelope.clone());

            let activity_id = ActivityId::from_sequence_position(item.ordinal);
            events.push(Event::ActivityScheduled {
                envelope: scheduled_envelope,
                activity_id: activity_id.clone(),
                activity_type: item.activity_type.clone(),
                input: item.input.clone(),
                // NSTQ-3: record the pool selector on the event from the same `FanOutItem`
                // (NSTQ-2) that stamps the outbox row, so the durable history is the
                // source-of-truth backstop for re-targeting this queue on recovery.
                task_queue: item.task_queue.clone(),
                // NODE-3: record the OPTIONAL node affinity on the event from the same `FanOutItem`
                // (NODE-2) that stamps the outbox row, so the durable history is the
                // source-of-truth backstop for re-targeting this node on recovery.
                node: item.node.clone(),
            });
            events.push(Event::ActivityStarted {
                envelope: started_envelope,
                activity_id,
                // NOI-0: the genuine one-based attempt this fan-out dispatch belongs to, from the
                // same `FanOutItem` that stamps the outbox row.
                attempt: item.attempt,
            });
            outbox_rows.push(
                OutboxRow::pending(
                    self.workflow_id.clone(),
                    item.ordinal,
                    item.activity_type.clone(),
                    item.input.clone(),
                    recorded_at,
                )
                .with_run_id(self.run_id.clone())
                .with_namespace(item.namespace.clone())
                .with_task_queue(item.task_queue.clone())
                .with_node(item.node.clone()),
            );
        }

        let expected_seq = self.sequence.current();
        self.store
            .append_with_outbox(
                self.write_token,
                &self.workflow_id,
                &events,
                expected_seq,
                &outbox_rows,
            )
            .await?;
        self.sequence.mark_append_success(events.len())
    }

    /// Re-arms the durable outbox rows for `items` back to claimable `Pending` (crash-recovery
    /// re-stage).
    ///
    /// On first arrival after a restart, an activity whose `ActivityScheduled` is recorded but which
    /// has no terminal event lost its in-flight dispatch when the previous engine process died. This
    /// re-stages that dispatch by flipping each ordinal's outbox row back to `Pending` so the
    /// `OutboxDispatcher` re-dispatches it, instead of driving an in-process completion task.
    ///
    /// Each [`FanOutItem`] maps to a fresh [`OutboxRow::pending`] exactly as
    /// [`Recorder::record_fan_out_dispatch`] does (same `dispatch_key = "{workflow_id}:{ordinal}"`),
    /// so the re-arm targets the row the original dispatch staged. This is a **pure outbox-table op**:
    /// it writes no history events and does NOT touch the sequence head — re-dispatch is at-least-once
    /// and the completion dedup ([`Recorder::record_fan_out_completion`]) absorbs any redelivery. An
    /// empty `items` slice is a no-op.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if the store rejects the re-arm (including an outbox-unaware
    /// backend refusing a non-empty re-arm).
    pub async fn rearm_outbox_pending(
        &self,
        recorded_at: DateTime<Utc>,
        items: &[FanOutItem],
    ) -> Result<(), DurabilityError> {
        if items.is_empty() {
            return Ok(());
        }

        let rows: Vec<OutboxRow> = items
            .iter()
            .map(|item| {
                OutboxRow::pending(
                    self.workflow_id.clone(),
                    item.ordinal,
                    item.activity_type.clone(),
                    item.input.clone(),
                    recorded_at,
                )
                .with_run_id(self.run_id.clone())
                .with_namespace(item.namespace.clone())
                .with_task_queue(item.task_queue.clone())
                .with_node(item.node.clone())
            })
            .collect();
        self.store.rearm_outbox_pending(&rows).await?;
        Ok(())
    }

    /// Store-backed completion dedup for one fan-out ordinal: the cross-node completion chokepoint.
    ///
    /// Determines whether [`ActivityId::from_sequence_position(ordinal)`](ActivityId::from_sequence_position)
    /// already has a terminal ([`Event::ActivityCompleted`] or [`Event::ActivityFailed`]) in this
    /// workflow's recorded history. This is the same "resolved" predicate the live dispatch path uses
    /// (`nif_collect.rs::recorded_terminal`): an ordinal is resolved once either terminal is present.
    ///
    /// - If the ordinal is **already resolved**, returns [`FanOutCompletionResult::Dropped`] WITHOUT
    ///   writing anything — no store append, the sequence head is unchanged. This is the core dedup
    ///   invariant: a duplicate completion (e.g. a redelivered cross-node send) never appends a
    ///   second terminal.
    /// - If the ordinal is **not yet resolved**, appends the single terminal event derived from
    ///   `outcome` ([`Event::ActivityCompleted`] for [`FanOutOutcome::Completed`],
    ///   [`Event::ActivityFailed`] for [`FanOutOutcome::Failed`]) through the normal terminal-append
    ///   path; the head advances by exactly one and it returns [`FanOutCompletionResult::Recorded`].
    ///
    /// # Boundary
    ///
    /// This method is **additive** and does NOT touch the runtime completion sink: it does not wake
    /// any workflow PID, does not signal `WorkerActivityDispatcher`, and does not interact with
    /// `nif_collect.rs`. Routing a recorded completion to a waiting collect is the later flag-gated
    /// cutover's responsibility; this method only owns the durable dedup-and-append.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError`] if reading history fails, if the terminal append is rejected (a
    /// [`StoreError::SequenceConflict`](aion_store::StoreError::SequenceConflict) surfaces as a hard
    /// error with the head unadvanced, mirroring the single-writer discipline), or if the sequence
    /// tracker cannot advance after a successful append.
    pub async fn record_fan_out_completion(
        &mut self,
        recorded_at: DateTime<Utc>,
        ordinal: u64,
        run_id: Option<RunId>,
        outcome: FanOutOutcome,
    ) -> Result<FanOutCompletionResult, DurabilityError> {
        let activity_id = ActivityId::from_sequence_position(ordinal);
        let history = self.store.read_history(&self.workflow_id).await?;
        if run_id_mismatches_recorder(run_id.as_ref(), self.run_id.as_ref()) {
            return Ok(FanOutCompletionResult::Dropped);
        }
        if ordinal_is_resolved(completion_history(&history, run_id.as_ref()), &activity_id) {
            return Ok(FanOutCompletionResult::Dropped);
        }

        match outcome {
            FanOutOutcome::Completed { result, attempt } => {
                self.append_with(recorded_at, |envelope| Event::ActivityCompleted {
                    envelope,
                    activity_id,
                    result,
                    // NOI-0: the genuine one-based attempt that produced this completion.
                    attempt,
                })
                .await?;
            }
            FanOutOutcome::Failed { error, attempt } => {
                self.append_with(recorded_at, |envelope| Event::ActivityFailed {
                    envelope,
                    activity_id,
                    error,
                    attempt,
                })
                .await?;
            }
        }
        Ok(FanOutCompletionResult::Recorded)
    }
}

fn run_id_mismatches_recorder(completion: Option<&RunId>, current: Option<&RunId>) -> bool {
    matches!((completion, current), (Some(completion), Some(current)) if completion != current)
}

fn completion_history<'a>(history: &'a [Event], run_id: Option<&RunId>) -> &'a [Event] {
    match run_id {
        Some(run_id) => run_segment(history, run_id),
        None => history,
    }
}

/// Whether `activity_id`'s ordinal already has a terminal ([`Event::ActivityCompleted`],
/// [`Event::ActivityFailed`], or [`Event::ActivityCancelled`]) recorded in `history`.
///
/// Mirrors the full "resolved" terminal set of `nif_collect.rs::recorded_terminal` — including
/// `ActivityCancelled`. A cancelled ordinal IS terminally resolved, so a late worker completion
/// arriving for it must be dropped, not recorded over the cancellation.
fn ordinal_is_resolved(history: &[Event], activity_id: &ActivityId) -> bool {
    history.iter().any(|event| match event {
        Event::ActivityCompleted {
            activity_id: recorded,
            ..
        }
        | Event::ActivityFailed {
            activity_id: recorded,
            ..
        }
        | Event::ActivityCancelled {
            activity_id: recorded,
            ..
        } => recorded == activity_id,
        _ => false,
    })
}