aion-core 0.31.0

//! Which worker holds a run's open attempts, folded from its own history.
//!
//! Exactly like [`crate::WorkflowStatus`] and [`crate::current_step`], the
//! answer is a PROJECTION of the authoritative event history. There is one
//! transition function, [`apply_lease_transition`], and every reader uses it:
//! the whole-history fold ([`outstanding_leases`]) and the engine's
//! per-append visibility touch apply the same function to the same events,
//! so a stored row and a summary built from history cannot disagree about
//! who is working on what.
//!
//! # The fold, precisely
//!
//! Only the CURRENT lease segment counts — everything after the last
//! [`Event::WorkflowStarted`] or [`Event::WorkflowReopened`]. A reopen
//! supersedes the terminals of the activities it names and re-dispatches
//! them, so a lease recorded before it belongs to a delivery that is no
//! longer the one in flight; the next delivery records its own.
//!
//! - [`Event::ActivityLeased`] for `(activity, attempt)` REPLACES any earlier
//!   lease of the same attempt and moves it to the end of the list. The
//!   transport is at-least-once, so the same attempt can be delivered twice;
//!   the LAST lease is the worker that holds it now.
//! - `ActivityCompleted` / `ActivityFailed` / `ActivityCancelled` for
//!   `(activity, attempt)` remove that attempt's lease. There is no
//!   activity-level timeout event; an attempt that times out is recorded as
//!   `ActivityFailed`.
//! - Every other event leaves the list as it was — including
//!   [`Event::ActivityStarted`], which is recorded at DISPATCH, before any
//!   worker holds the work (WA-010 §2.5), and
//!   [`Event::ActivityAdoptionOffered`], which re-offers an attempt the
//!   worker may still hold.
//!
//! `current_worker` is the worker of the LAST outstanding lease: the attempt
//! most recently taken. When two attempts are outstanding and the later one
//! terminates, the earlier one is shown again — it is still being worked.
//!
//! # What "absent" means
//!
//! An empty list (and a `None` current worker) is UNATTRIBUTED, and that word
//! covers two different facts a reader must not conflate: a history recorded
//! before lease events existed, and an install that knew the worker and
//! failed to record it. History alone cannot tell them apart, which is why
//! every read surface also carries the install's
//! [`crate::ReadProvenance::lease_record_failures_total`].

use serde::{Deserialize, Serialize};

use crate::{ActivityId, Event, WorkerAttribution};

/// One attempt a worker holds: the LAST lease recorded for it with no
/// terminal since.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Debug, PartialEq, Eq)]
pub struct OutstandingLease {
    /// The activity ordinal the attempt belongs to.
    pub activity_id: ActivityId,
    /// One-based attempt number — the same `attempt` its `ActivityStarted`
    /// and `ActivityLeased` carry.
    pub attempt: u32,
    /// The worker that holds it, by the durable names its lease recorded.
    pub worker: WorkerAttribution,
}

/// Applies ONE event to a list of outstanding leases, in place.
///
/// This is the single transition every reader shares — see the module
/// documentation for the rules. Events that say nothing about leases leave
/// the list untouched. Callers folding a whole history are responsible for
/// resetting the list at a segment boundary; this function does not, so the
/// engine's per-append touch (which re-projects the whole row on lifecycle
/// events) can apply it to a stored row event by event.
pub fn apply_lease_transition(leases: &mut Vec<OutstandingLease>, event: &Event) {
    match event {
        Event::ActivityLeased {
            activity_id,
            attempt,
            worker,
            ..
        } => {
            leases
                .retain(|lease| !(lease.activity_id == *activity_id && lease.attempt == *attempt));
            leases.push(OutstandingLease {
                activity_id: activity_id.clone(),
                attempt: *attempt,
                worker: worker.clone(),
            });
        }
        Event::ActivityCompleted {
            activity_id,
            attempt,
            ..
        }
        | Event::ActivityFailed {
            activity_id,
            attempt,
            ..
        }
        | Event::ActivityCancelled {
            activity_id,
            attempt,
            ..
        } => {
            leases
                .retain(|lease| !(lease.activity_id == *activity_id && lease.attempt == *attempt));
        }
        _ => {}
    }
}

/// Every attempt a worker currently holds in `events`' current lease
/// segment, in the order the leases were recorded (earliest first).
#[must_use]
pub fn outstanding_leases(events: &[Event]) -> Vec<OutstandingLease> {
    let segment_start = events
        .iter()
        .rposition(|event| {
            matches!(
                event,
                Event::WorkflowStarted { .. } | Event::WorkflowReopened { .. }
            )
        })
        .map_or(0, |index| index + 1);
    let mut leases = Vec::new();
    for event in &events[segment_start..] {
        apply_lease_transition(&mut leases, event);
    }
    leases
}

/// What a run's WHOLE history says about lease recording, as counts a
/// reader can hold an unattributed attempt against (WA-010 R4 follow-up).
///
/// A console over a windowed history cannot tell "no lease in view" from
/// "no lease ever", and no fold can tell "recorded before lease events
/// existed" from "never recorded" — the history carries the package version a
/// run started under, not the server version. So the server states the two
/// things it can count over the full history, and the reader says "no lease
/// recorded across N dispatched attempts" rather than guessing why.
#[derive(Serialize, Deserialize, ts_rs::TS, Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct LeaseRecording {
    /// `ActivityLeased` events in the whole history.
    pub leases_recorded: u64,
    /// `ActivityStarted` events in the whole history — every dispatched
    /// attempt, whether or not a lease was recorded for it.
    pub dispatched_attempts: u64,
    /// Distinct `(activity, attempt)` keys with at least one lease. The
    /// transport is at-least-once, so `leases_recorded` can exceed this;
    /// "every attempt was attributed" is `attempts_with_a_lease ==
    /// dispatched_attempts`, never a comparison against `leases_recorded`.
    pub attempts_with_a_lease: u64,
}

/// Counts leases and dispatched attempts over the WHOLE history, every
/// segment — a statement of the record, not of the current segment.
#[must_use]
pub fn lease_recording(events: &[Event]) -> LeaseRecording {
    let mut counts = LeaseRecording::default();
    let mut attributed: std::collections::HashSet<(&ActivityId, u32)> =
        std::collections::HashSet::new();
    for event in events {
        match event {
            Event::ActivityLeased {
                activity_id,
                attempt,
                ..
            } => {
                counts.leases_recorded += 1;
                if attributed.insert((activity_id, *attempt)) {
                    counts.attempts_with_a_lease += 1;
                }
            }
            Event::ActivityStarted { .. } => counts.dispatched_attempts += 1,
            _ => {}
        }
    }
    counts
}

/// The worker holding the run's most recently leased open attempt, or `None`
/// when the run is UNATTRIBUTED (no outstanding lease — see the module
/// documentation for what that does and does not mean).
#[must_use]
pub fn current_worker(leases: &[OutstandingLease]) -> Option<WorkerAttribution> {
    leases.last().map(|lease| lease.worker.clone())
}

#[cfg(test)]
mod tests {
    use chrono::{DateTime, Utc};

    use super::{OutstandingLease, apply_lease_transition, current_worker, outstanding_leases};
    use crate::{
        ActivityError, ActivityErrorKind, ActivityId, Event, EventEnvelope, Payload, RunId,
        WorkerAttribution, WorkerTransport, WorkflowId,
    };

    fn workflow_id() -> WorkflowId {
        WorkflowId::new(uuid::Uuid::from_u128(7))
    }

    fn envelope(seq: u64) -> EventEnvelope {
        EventEnvelope {
            seq,
            recorded_at: DateTime::<Utc>::from_timestamp(
                1_700_000_000 + i64::try_from(seq).unwrap_or(0),
                0,
            )
            .unwrap_or_default(),
            workflow_id: workflow_id(),
        }
    }

    fn worker(identity: &str) -> WorkerAttribution {
        WorkerAttribution {
            identity: identity.to_owned(),
            task_queue: String::from("billing"),
            node: Some(String::from("n1")),
            deployment: None,
            instance_id: None,
            transport: WorkerTransport::Grpc,
        }
    }

    fn started(seq: u64) -> Result<Event, Box<dyn std::error::Error>> {
        Ok(Event::WorkflowStarted {
            envelope: envelope(seq),
            workflow_type: String::from("checkout"),
            input: Payload::from_json(&serde_json::json!({}))?,
            run_id: RunId::new(uuid::Uuid::from_u128(1)),
            parent_run_id: None,
            parent_workflow_id: None,
            package_version: crate::PackageVersion::new("a".repeat(64)),
        })
    }

    fn leased(seq: u64, activity: u64, attempt: u32, identity: &str) -> Event {
        Event::ActivityLeased {
            envelope: envelope(seq),
            activity_id: ActivityId::from_sequence_position(activity),
            attempt,
            worker: worker(identity),
        }
    }

    fn completed(
        seq: u64,
        activity: u64,
        attempt: u32,
    ) -> Result<Event, Box<dyn std::error::Error>> {
        Ok(Event::ActivityCompleted {
            envelope: envelope(seq),
            activity_id: ActivityId::from_sequence_position(activity),
            attempt,
            result: Payload::from_json(&serde_json::json!({}))?,
        })
    }

    fn failed(seq: u64, activity: u64, attempt: u32) -> Event {
        Event::ActivityFailed {
            envelope: envelope(seq),
            activity_id: ActivityId::from_sequence_position(activity),
            attempt,
            error: ActivityError {
                kind: ActivityErrorKind::Terminal,
                message: String::from("boom"),
                details: None,
            },
        }
    }

    fn identities(leases: &[OutstandingLease]) -> Vec<(u64, u32, &str)> {
        leases
            .iter()
            .map(|lease| {
                (
                    lease.activity_id.sequence_position(),
                    lease.attempt,
                    lease.worker.identity.as_str(),
                )
            })
            .collect()
    }

    #[test]
    fn a_lease_sets_the_current_worker_and_its_terminal_clears_it()
    -> Result<(), Box<dyn std::error::Error>> {
        let mut history = vec![started(1)?, leased(2, 2, 1, "w-a")];
        let leases = outstanding_leases(&history);
        assert_eq!(identities(&leases), vec![(2, 1, "w-a")]);
        assert_eq!(
            current_worker(&leases).map(|w| w.identity),
            Some(String::from("w-a"))
        );

        history.push(completed(3, 2, 1)?);
        let leases = outstanding_leases(&history);
        assert!(
            leases.is_empty(),
            "the completion clears the attempt's lease"
        );
        assert_eq!(current_worker(&leases), None);
        Ok(())
    }

    #[test]
    fn a_second_lease_of_the_same_attempt_replaces_the_first()
    -> Result<(), Box<dyn std::error::Error>> {
        let history = vec![started(1)?, leased(2, 2, 1, "w-a"), leased(3, 2, 1, "w-b")];
        let leases = outstanding_leases(&history);
        assert_eq!(
            identities(&leases),
            vec![(2, 1, "w-b")],
            "last lease per attempt wins"
        );
        Ok(())
    }

    #[test]
    fn a_terminal_for_another_attempt_leaves_the_lease_standing()
    -> Result<(), Box<dyn std::error::Error>> {
        // Attempt 1 failed after attempt 2 was leased (a late terminal from a
        // retried delivery): attempt 2's lease is untouched.
        let history = vec![started(1)?, leased(2, 2, 2, "w-b"), failed(3, 2, 1)];
        assert_eq!(
            identities(&outstanding_leases(&history)),
            vec![(2, 2, "w-b")]
        );
        Ok(())
    }

    #[test]
    fn when_the_latest_attempt_terminates_the_earlier_outstanding_one_is_shown()
    -> Result<(), Box<dyn std::error::Error>> {
        let history = vec![
            started(1)?,
            leased(2, 2, 1, "w-a"),
            leased(3, 3, 1, "w-b"),
            completed(4, 3, 1)?,
        ];
        let leases = outstanding_leases(&history);
        assert_eq!(identities(&leases), vec![(2, 1, "w-a")]);
        assert_eq!(
            current_worker(&leases).map(|w| w.identity),
            Some(String::from("w-a"))
        );
        Ok(())
    }

    #[test]
    fn a_history_with_no_lease_is_unattributed() -> Result<(), Box<dyn std::error::Error>> {
        let history = vec![started(1)?];
        assert!(outstanding_leases(&history).is_empty());
        assert_eq!(current_worker(&[]), None);
        Ok(())
    }

    #[test]
    fn lease_recording_counts_the_whole_history_across_segments()
    -> Result<(), Box<dyn std::error::Error>> {
        let started_attempt = |seq: u64, activity: u64| Event::ActivityStarted {
            envelope: envelope(seq),
            activity_id: ActivityId::from_sequence_position(activity),
            attempt: 1,
        };
        let history = vec![
            started(1)?,
            started_attempt(2, 2),
            leased(3, 2, 1, "w-a"),
            Event::WorkflowReopened {
                envelope: envelope(4),
                run_id: RunId::new(uuid::Uuid::from_u128(1)),
                reopened: vec![ActivityId::from_sequence_position(2)],
            },
            started_attempt(5, 2),
        ];
        let counts = super::lease_recording(&history);
        assert_eq!(
            counts.leases_recorded, 1,
            "the lease before the reopen still counts"
        );
        assert_eq!(counts.dispatched_attempts, 2, "both dispatches count");
        assert_eq!(counts.attempts_with_a_lease, 1);
        // At-least-once: a second lease for the SAME attempt is a lease
        // recorded, not another attempt attributed.
        let mut redelivered = history.clone();
        redelivered.push(leased(6, 2, 1, "w-b"));
        let counts = super::lease_recording(&redelivered);
        assert_eq!(
            (counts.leases_recorded, counts.attempts_with_a_lease),
            (2, 1)
        );
        assert!(
            super::outstanding_leases(&history).is_empty(),
            "while the current segment's attribution is empty"
        );
        Ok(())
    }

    #[test]
    fn a_reopen_starts_a_fresh_segment() -> Result<(), Box<dyn std::error::Error>> {
        let history = vec![
            started(1)?,
            leased(2, 2, 1, "w-a"),
            Event::WorkflowReopened {
                envelope: envelope(3),
                run_id: RunId::new(uuid::Uuid::from_u128(1)),
                reopened: vec![ActivityId::from_sequence_position(2)],
            },
        ];
        assert!(
            outstanding_leases(&history).is_empty(),
            "a lease recorded before the reopen belongs to a superseded delivery"
        );
        Ok(())
    }

    #[test]
    fn the_incremental_transition_matches_the_fold() -> Result<(), Box<dyn std::error::Error>> {
        let history = vec![
            started(1)?,
            leased(2, 2, 1, "w-a"),
            leased(3, 3, 1, "w-b"),
            leased(4, 2, 1, "w-c"),
            failed(5, 3, 1),
        ];
        let mut incremental = Vec::new();
        for event in &history[1..] {
            apply_lease_transition(&mut incremental, event);
        }
        assert_eq!(incremental, outstanding_leases(&history));
        assert_eq!(identities(&incremental), vec![(2, 1, "w-c")]);
        Ok(())
    }
}