behavior_actors/operations/

health.rs

//! Versioned component-health aggregation.

#[cfg(test)]
use behavior::Recipient;
use behavior::{Actions, Address, Behavior, BehaviorActed, BehaviorBase, Never, NoBirths, User};
use thiserror::Error;

use crate::DeliveryRoute;

/// Monotonic version of one component's health evidence.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct ObservationVersion(pub u64);

/// Closed health classification ordered from best to worst.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum HealthStatus {
    /// The component is fully available under its declared policy.
    Healthy,
    /// The component remains usable with a declared impairment.
    Degraded,
    /// The component is unavailable or must not receive work.
    Unhealthy,
}

/// One present component in a [`HealthReport`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ComponentHealth<K> {
    /// Application-defined component identity.
    pub component: K,
    /// Latest committed evidence version.
    pub version: ObservationVersion,
    /// Latest committed classification.
    pub status: HealthStatus,
}

/// Complete retained state for one known component.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ComponentHealthState<K> {
    /// Latest evidence describes a present component.
    Present(ComponentHealth<K>),
    /// The component was explicitly removed at this version.
    Removed {
        /// Application-defined component identity.
        component: K,
        /// Removal version retained to reject stale resurrection.
        version: ObservationVersion,
    },
}

impl<K> ComponentHealthState<K> {
    fn component(&self) -> &K {
        match self {
            Self::Present(health) => &health.component,
            Self::Removed { component, .. } => component,
        }
    }

    const fn version(&self) -> ObservationVersion {
        match self {
            Self::Present(health) => health.version,
            Self::Removed { version, .. } => *version,
        }
    }
}

/// Factual point-in-time result returned by [`Health`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HealthReport<K> {
    /// Present components in first-observation order; tombstones are omitted.
    pub components: Vec<ComponentHealth<K>>,
}

impl<K> HealthReport<K> {
    /// Worst status among present components, or `Healthy` when none exist.
    #[must_use]
    pub fn overall(&self) -> HealthStatus {
        self.components
            .iter()
            .map(|component| component.status)
            .max()
            .unwrap_or(HealthStatus::Healthy)
    }
}

/// Commands accepted by [`Health`].
pub enum HealthMessage<K, Route> {
    /// Commit versioned evidence for a present component.
    Observe {
        /// Component identity.
        component: K,
        /// Evidence version within that component's stream.
        version: ObservationVersion,
        /// New factual classification.
        status: HealthStatus,
    },
    /// Commit a versioned tombstone.
    Remove {
        /// Component identity.
        component: K,
        /// Removal version within that component's stream.
        version: ObservationVersion,
    },
    /// Return a point-in-time report to a typed recipient.
    Query {
        /// Recipient whose protocol accepts [`HealthReport<K>`].
        reply_to: Route,
    },
}

/// Rejected versioned health evidence.
#[derive(Debug, Error, Clone, PartialEq, Eq)]
pub enum HealthError<K> {
    /// Evidence predates the latest committed version.
    #[error("health evidence is stale")]
    Stale {
        /// Component whose evidence was rejected.
        component: K,
        /// Rejected version.
        observed: ObservationVersion,
        /// Latest committed version.
        current: ObservationVersion,
        /// Exact rejected evidence.
        evidence: HealthEvidence,
    },
    /// Evidence reuses a committed version with different meaning.
    #[error("health evidence contradicts the committed value at the same version")]
    ConflictingVersion {
        /// Component whose evidence was rejected.
        component: K,
        /// Reused version.
        version: ObservationVersion,
        /// Exact rejected evidence.
        evidence: HealthEvidence,
    },
}

/// Exact meaning of one submitted health fact.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum HealthEvidence {
    /// The component is present with this classification.
    Present(HealthStatus),
    /// The component is removed at the submitted version.
    Removed,
}

/// Versioned typed health aggregation behavior.
///
/// State is an insertion-ordered product of [`ComponentHealthState`] values.
/// A greater version replaces the prior state. An identical observation is
/// idempotent. A lower version is [`HealthError::Stale`], and a same-version
/// contradiction is [`HealthError::ConflictingVersion`]; both preserve the
/// complete prior state. Removal retains a tombstone, so stale evidence cannot
/// resurrect a component. Query emits one named typed delivery and never
/// mutates state. Initialization is empty, the behavior never terminates by
/// policy, and it requires only ordinary typed delivery interpretation.
/// Versioning, aggregate ordering, and empty-set health are Bombay policy, not
/// actor-model laws. No method has a semantic panic condition.
pub struct Health<A, K, Route>
where
    A: Address,
    Route: DeliveryRoute<Protocol: behavior::Protocol<Addr = A, Msg = HealthReport<K>>>,
{
    components: Vec<ComponentHealthState<K>>,
    address: core::marker::PhantomData<fn() -> (A, Route)>,
}

impl<A, K, Route> Health<A, K, Route>
where
    A: Address,
    Route: DeliveryRoute<Protocol: behavior::Protocol<Addr = A, Msg = HealthReport<K>>>,
{
    /// Construct an empty health definition.
    #[must_use]
    pub const fn new() -> Self {
        Self {
            components: Vec::new(),
            address: core::marker::PhantomData,
        }
    }

    /// Borrow every retained present or removed component state.
    #[must_use]
    pub fn components(&self) -> &[ComponentHealthState<K>] {
        &self.components
    }
}

impl<A, K, Route> Default for Health<A, K, Route>
where
    A: Address,
    Route: DeliveryRoute<Protocol: behavior::Protocol<Addr = A, Msg = HealthReport<K>>>,
{
    fn default() -> Self {
        Self::new()
    }
}

impl<A, K, Route> BehaviorBase for Health<A, K, Route>
where
    A: Address,
    Route: DeliveryRoute<Protocol: behavior::Protocol<Addr = A, Msg = HealthReport<K>>>,
{
    type Base = Self;

    fn base(&self) -> &Self {
        self
    }
}

impl<A, K, Route> Health<A, K, Route>
where
    A: Address,
    K: Clone + Eq,
    Route: DeliveryRoute<Protocol: behavior::Protocol<Addr = A, Msg = HealthReport<K>>>,
{
    fn commit(
        &mut self,
        component: K,
        version: ObservationVersion,
        evidence: HealthEvidence,
    ) -> Result<(), HealthError<K>> {
        let replacement = match evidence {
            HealthEvidence::Present(status) => ComponentHealthState::Present(ComponentHealth {
                component: component.clone(),
                version,
                status,
            }),
            HealthEvidence::Removed => ComponentHealthState::Removed {
                component: component.clone(),
                version,
            },
        };
        let Some(index) = self
            .components
            .iter()
            .position(|state| state.component() == &component)
        else {
            self.components.push(replacement);
            return Ok(());
        };
        let current = self.components[index].version();
        if version < current {
            return Err(HealthError::Stale {
                component,
                observed: version,
                current,
                evidence,
            });
        }
        if version == current {
            if self.components[index] == replacement {
                return Ok(());
            }
            return Err(HealthError::ConflictingVersion {
                component,
                version,
                evidence,
            });
        }
        self.components[index] = replacement;
        Ok(())
    }
    fn report(&self) -> HealthReport<K> {
        let components = self
            .components
            .iter()
            .filter_map(|state| match state {
                ComponentHealthState::Present(health) => Some(health.clone()),
                ComponentHealthState::Removed { .. } => None,
            })
            .collect::<Vec<_>>();
        HealthReport { components }
    }
}

impl<A, K, Route> behavior::Protocol for Health<A, K, Route>
where
    A: Address,
    Route: DeliveryRoute<Protocol: behavior::Protocol<Addr = A, Msg = HealthReport<K>>>,
{
    type Addr = A;
    type Msg = HealthMessage<K, Route>;
}

impl<A, K, Route> Behavior for Health<A, K, Route>
where
    A: Address,
    K: Clone + Eq,
    Route: DeliveryRoute<Protocol: behavior::Protocol<Addr = A, Msg = HealthReport<K>>>,
    Route::Sends: behavior::SendsFor<User<A, HealthMessage<K, Route>>>,
{
    type Protocol = Self;
    type Event = User<A, behavior::BehaviorMessage<Self>>;
    type Sends = Route::Sends;
    type Ph = Never;
    type Error = HealthError<K>;
    type Birth = NoBirths;

    fn transition(&mut self, _: behavior::ActiveTurn, event: Self::Event) -> BehaviorActed<Self> {
        match event.message {
            HealthMessage::Observe {
                component,
                version,
                status,
            } => {
                self.commit(component, version, HealthEvidence::Present(status))?;
                Ok(Actions::cont())
            }
            HealthMessage::Remove { component, version } => {
                self.commit(component, version, HealthEvidence::Removed)?;
                Ok(Actions::cont())
            }
            HealthMessage::Query { reply_to } => Ok(Actions::send(reply_to.deliver(self.report()))),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::Activate as _;
    use behavior::MailAddr;

    struct Reply;

    impl behavior::Protocol for Reply {
        type Addr = MailAddr;
        type Msg = HealthReport<u8>;
    }

    impl Behavior for Reply {
        type Protocol = Self;
        type Event = User<MailAddr, behavior::BehaviorMessage<Self>>;
        type Sends = Vec<Never>;
        type Ph = Never;
        type Error = Never;
        type Birth = NoBirths;

        fn transition(&mut self, _: behavior::ActiveTurn, _: Self::Event) -> BehaviorActed<Self> {
            Ok(Actions::cont())
        }
    }

    type TestHealth = Health<MailAddr, u8, Recipient<Reply>>;

    #[test]
    fn stale_and_conflicting_evidence_preserve_committed_state() {
        let mut health = (TestHealth::new()).initialize().unwrap().behavior;
        let observed = health
            .receive(
                MailAddr(9),
                HealthMessage::Observe {
                    component: 1,
                    version: ObservationVersion(3),
                    status: HealthStatus::Degraded,
                },
            )
            .unwrap();
        assert!(observed.sends.is_empty());
        assert!(observed.creates.is_empty());
        assert_eq!(observed.become_, behavior::Step::Continue);

        let rejection = health.receive(
            MailAddr(9),
            HealthMessage::Observe {
                component: 1,
                version: ObservationVersion(2),
                status: HealthStatus::Healthy,
            },
        );
        assert!(matches!(
            rejection,
            Err(HealthError::Stale {
                component: 1,
                observed: ObservationVersion(2),
                current: ObservationVersion(3),
                evidence: HealthEvidence::Present(HealthStatus::Healthy),
            })
        ));
        let rejection = health.receive(
            MailAddr(9),
            HealthMessage::Observe {
                component: 1,
                version: ObservationVersion(3),
                status: HealthStatus::Unhealthy,
            },
        );
        assert!(matches!(
            rejection,
            Err(HealthError::ConflictingVersion {
                component: 1,
                version: ObservationVersion(3),
                evidence: HealthEvidence::Present(HealthStatus::Unhealthy),
            })
        ));
        assert_eq!(
            health.components(),
            [ComponentHealthState::Present(ComponentHealth {
                component: 1,
                version: ObservationVersion(3),
                status: HealthStatus::Degraded,
            })]
        );
    }

    #[test]
    fn tombstone_rejects_resurrection_and_report_aggregates_worst_status() {
        let reply = Recipient::<Reply>::global(MailAddr(8));
        let mut health = (TestHealth::new()).initialize().unwrap().behavior;
        for (component, status) in [(1, HealthStatus::Healthy), (2, HealthStatus::Unhealthy)] {
            let observed = health
                .receive(
                    MailAddr(9),
                    HealthMessage::Observe {
                        component,
                        version: ObservationVersion(1),
                        status,
                    },
                )
                .unwrap();
            assert!(observed.sends.is_empty());
            assert!(observed.creates.is_empty());
            assert_eq!(observed.become_, behavior::Step::Continue);
        }
        let report = health
            .receive(MailAddr(9), HealthMessage::Query { reply_to: reply })
            .unwrap();
        let sent = report
            .sends
            .into_iter()
            .next()
            .expect("one report delivery");
        assert_eq!(sent.to, reply);
        assert_eq!(sent.message.overall(), HealthStatus::Unhealthy);
        assert_eq!(
            sent.message.components,
            vec![
                ComponentHealth {
                    component: 1,
                    version: ObservationVersion(1),
                    status: HealthStatus::Healthy,
                },
                ComponentHealth {
                    component: 2,
                    version: ObservationVersion(1),
                    status: HealthStatus::Unhealthy,
                },
            ]
        );

        let removed = health
            .receive(
                MailAddr(9),
                HealthMessage::Remove {
                    component: 2,
                    version: ObservationVersion(2),
                },
            )
            .unwrap();
        assert!(removed.sends.is_empty());
        assert!(removed.creates.is_empty());
        assert_eq!(removed.become_, behavior::Step::Continue);
        let rejection = health.receive(
            MailAddr(9),
            HealthMessage::Observe {
                component: 2,
                version: ObservationVersion(1),
                status: HealthStatus::Healthy,
            },
        );
        assert!(matches!(rejection, Err(HealthError::Stale { .. })));
        let after = health
            .receive(MailAddr(9), HealthMessage::Query { reply_to: reply })
            .unwrap();
        let sent = after.sends.into_iter().next().expect("one report delivery");
        assert_eq!(sent.to, reply);
        assert_eq!(sent.message.overall(), HealthStatus::Healthy);
        assert_eq!(
            sent.message.components,
            vec![ComponentHealth {
                component: 1,
                version: ObservationVersion(1),
                status: HealthStatus::Healthy,
            }]
        );
    }
}