frame_core/

event.rs

//! Typed, bounded lifecycle event delivery.

use std::collections::VecDeque;
use std::num::NonZeroUsize;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Condvar, Mutex, Weak};
use std::time::Duration;

use serde::{Deserialize, Serialize};
use thiserror::Error;

use crate::capability::CapabilityDenied;
use crate::component::ComponentId;

/// An externally observable component lifecycle state.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub enum LifecycleState {
    /// Definition is registered but has not started.
    Registered,
    /// Module loading and child liveness checks are in progress.
    Starting,
    /// Every declared child completed a mailbox round-trip.
    Running,
    /// Ordered child and supervisor drain is in progress.
    Stopping,
    /// All component processes have normal tombstones.
    Stopped,
    /// Start or supervision failed; the associated status carries the reason.
    Failed,
    /// Modules and registration were removed without residue.
    Removed,
}

/// One ordered item on the component lifecycle and security-news stream.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub struct LifecycleEvent {
    /// Registry-wide monotonic sequence number.
    pub sequence: u64,
    /// Component associated with this news item.
    pub component_id: ComponentId,
    /// Typed transition or capability denial payload.
    pub kind: LifecycleEventKind,
}

/// The typed payload carried by one lifecycle stream item.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
pub enum LifecycleEventKind {
    /// One component lifecycle transition.
    Transition {
        /// State before this transition, absent for initial registration.
        from: Option<LifecycleState>,
        /// State after this transition.
        to: LifecycleState,
    },
    /// One consuming act was denied by the fresh capability check.
    CapabilityDenied(CapabilityDenied),
    /// One committed live fragment content update (F-5a R2/R3). The stream
    /// carries the news, never the bytes: the registry's fragment snapshot
    /// stays the authority, and a consumer that misses this event converges
    /// through its next lag-forced resync.
    FragmentContentUpdated {
        /// The updated fragment.
        key: crate::fragment::FragmentKey,
    },
}

/// A bounded subscription that reports every event it had to discard.
pub struct LifecycleSubscription {
    queue: Arc<SubscriberQueue>,
}

impl LifecycleSubscription {
    /// Waits up to `timeout` for the next retained event.
    ///
    /// # Errors
    ///
    /// Returns a typed timeout, closure, or synchronization failure.
    pub fn recv_timeout(&self, timeout: Duration) -> Result<LifecycleEvent, EventReceiveError> {
        let guard = self
            .queue
            .state
            .lock()
            .map_err(|_| EventReceiveError::Poisoned)?;
        let (mut guard, _wait) = self
            .queue
            .ready
            .wait_timeout_while(guard, timeout, |state| {
                state.events.is_empty() && !state.closed
            })
            .map_err(|_| EventReceiveError::Poisoned)?;
        if let Some(event) = guard.events.pop_front() {
            return Ok(event);
        }
        if guard.closed {
            Err(EventReceiveError::Closed)
        } else {
            Err(EventReceiveError::Timeout)
        }
    }

    /// Returns the next retained event without waiting.
    ///
    /// # Errors
    ///
    /// Returns closure, emptiness, or synchronization failures distinctly.
    pub fn try_recv(&self) -> Result<LifecycleEvent, EventTryReceiveError> {
        let mut state = self
            .queue
            .state
            .lock()
            .map_err(|_| EventTryReceiveError::Poisoned)?;
        if let Some(event) = state.events.pop_front() {
            Ok(event)
        } else if state.closed {
            Err(EventTryReceiveError::Closed)
        } else {
            Err(EventTryReceiveError::Empty)
        }
    }

    /// Blocks until the next retained event arrives or the subscription is
    /// closed — the timeout-less sibling of [`Self::recv_timeout`], same
    /// condvar, no timer anywhere (Ruling B's sanctioned blocking shape:
    /// close is the only external wake, delivered by
    /// [`SubscriptionCloseHandle::close`] or this subscription's drop).
    ///
    /// # Errors
    ///
    /// Returns a typed closure or synchronization failure; never a timeout.
    pub fn recv(&self) -> Result<LifecycleEvent, EventReceiveError> {
        let guard = self
            .queue
            .state
            .lock()
            .map_err(|_| EventReceiveError::Poisoned)?;
        let mut guard = self
            .queue
            .ready
            .wait_while(guard, |state| state.events.is_empty() && !state.closed)
            .map_err(|_| EventReceiveError::Poisoned)?;
        if let Some(event) = guard.events.pop_front() {
            return Ok(event);
        }
        Err(EventReceiveError::Closed)
    }

    /// Returns the cumulative number of events dropped from this subscriber.
    #[must_use]
    pub fn lagged_events(&self) -> usize {
        self.queue.lagged.load(Ordering::Acquire)
    }

    /// Returns a handle that closes this subscription from another thread,
    /// waking every blocked receiver (the demand-driven teardown wake — the
    /// F-6a frame-core sanction amendment).
    ///
    /// The handle holds only a weak reference: retaining it never keeps the
    /// subscription's queue alive after the subscription drops.
    #[must_use]
    pub fn close_handle(&self) -> SubscriptionCloseHandle {
        SubscriptionCloseHandle {
            queue: Arc::downgrade(&self.queue),
        }
    }
}

/// Externally-holdable close switch for one [`LifecycleSubscription`].
pub struct SubscriptionCloseHandle {
    queue: Weak<SubscriberQueue>,
}

impl SubscriptionCloseHandle {
    /// Closes the subscription and wakes every blocked receiver. Idempotent;
    /// a no-op once the subscription itself has dropped.
    ///
    /// A poisoned subscriber lock cannot mark the queue closed, but blocked
    /// receivers are still woken and then observe their own typed
    /// [`EventReceiveError::Poisoned`] — the failure is propagated at the
    /// receive surface, never swallowed silently.
    pub fn close(&self) {
        if let Some(queue) = self.queue.upgrade() {
            if let Ok(mut state) = queue.state.lock() {
                state.closed = true;
            }
            queue.ready.notify_all();
        }
    }
}

impl Drop for LifecycleSubscription {
    fn drop(&mut self) {
        if let Ok(mut state) = self.queue.state.lock() {
            state.closed = true;
            self.queue.ready.notify_all();
        }
    }
}

/// Failure while waiting for an event.
#[derive(Clone, Copy, Debug, Error, Eq, PartialEq)]
pub enum EventReceiveError {
    /// No event arrived before the caller's deadline.
    #[error("lifecycle event receive timed out")]
    Timeout,
    /// The subscription was closed.
    #[error("lifecycle event subscription is closed")]
    Closed,
    /// Subscriber synchronization was poisoned by a panic.
    #[error("lifecycle event subscription synchronization is poisoned")]
    Poisoned,
}

/// Failure while polling for an event.
#[derive(Clone, Copy, Debug, Error, Eq, PartialEq)]
pub enum EventTryReceiveError {
    /// No retained event is currently available.
    #[error("lifecycle event subscription is empty")]
    Empty,
    /// The subscription was closed.
    #[error("lifecycle event subscription is closed")]
    Closed,
    /// Subscriber synchronization was poisoned by a panic.
    #[error("lifecycle event subscription synchronization is poisoned")]
    Poisoned,
}

#[derive(Default)]
pub(crate) struct EventHub {
    next_sequence: AtomicU64,
    subscribers: Mutex<Vec<Weak<SubscriberQueue>>>,
    /// Deterministic count of constructed events — the F-5a R6 zero-cost
    /// choke point: with no live subscriber, publication must return before
    /// ANY event is constructed, and this counter proves it stayed zero.
    #[cfg(test)]
    pub(crate) constructed: AtomicUsize,
}

impl EventHub {
    pub(crate) fn subscribe(
        &self,
        capacity: NonZeroUsize,
    ) -> Result<LifecycleSubscription, EventPublishError> {
        let queue = Arc::new(SubscriberQueue {
            capacity: capacity.get(),
            state: Mutex::new(QueueState::default()),
            ready: Condvar::new(),
            lagged: AtomicUsize::new(0),
        });
        self.subscribers
            .lock()
            .map_err(|_| EventPublishError::Poisoned)?
            .push(Arc::downgrade(&queue));
        Ok(LifecycleSubscription { queue })
    }

    pub(crate) fn publish_transition(
        &self,
        component_id: ComponentId,
        from: Option<LifecycleState>,
        to: LifecycleState,
    ) -> Result<(), EventPublishError> {
        self.publish(component_id, LifecycleEventKind::Transition { from, to })
    }

    pub(crate) fn publish_denial(&self, denial: CapabilityDenied) -> Result<(), EventPublishError> {
        self.publish(
            denial.component_id,
            LifecycleEventKind::CapabilityDenied(denial),
        )
    }

    pub(crate) fn publish_fragment_update(
        &self,
        key: crate::fragment::FragmentKey,
    ) -> Result<(), EventPublishError> {
        self.publish(
            key.component_id,
            LifecycleEventKind::FragmentContentUpdated { key },
        )
    }

    fn publish(
        &self,
        component_id: ComponentId,
        kind: LifecycleEventKind,
    ) -> Result<(), EventPublishError> {
        let mut subscribers = self
            .subscribers
            .lock()
            .map_err(|_| EventPublishError::Poisoned)?;
        subscribers.retain(|subscriber| subscriber.strong_count() > 0);
        if subscribers.is_empty() {
            return Ok(());
        }
        #[cfg(test)]
        self.constructed.fetch_add(1, Ordering::AcqRel);
        let event = LifecycleEvent {
            sequence: self.next_sequence.fetch_add(1, Ordering::AcqRel),
            component_id,
            kind,
        };
        for subscriber in subscribers.iter().filter_map(Weak::upgrade) {
            let mut state = subscriber
                .state
                .lock()
                .map_err(|_| EventPublishError::Poisoned)?;
            if state.events.len() == subscriber.capacity {
                let _discarded = state.events.pop_front();
                subscriber.lagged.fetch_add(1, Ordering::AcqRel);
            }
            state.events.push_back(event.clone());
            subscriber.ready.notify_one();
        }
        Ok(())
    }
}

#[derive(Debug, Error)]
pub(crate) enum EventPublishError {
    #[error("lifecycle event stream synchronization is poisoned")]
    Poisoned,
}

struct SubscriberQueue {
    capacity: usize,
    state: Mutex<QueueState>,
    ready: Condvar,
    lagged: AtomicUsize,
}

#[derive(Default)]
struct QueueState {
    events: VecDeque<LifecycleEvent>,
    closed: bool,
}

#[cfg(test)]
mod tests {
    #![allow(clippy::panic)]

    use std::num::NonZeroUsize;
    use std::sync::mpsc;
    use std::thread;
    use std::time::Duration;

    use super::{EventHub, EventReceiveError, LifecycleState};
    use crate::component::ComponentId;

    /// Generous wall for the TEST harness only (the code under test carries
    /// no timeout — that is the point).
    const HARNESS_WALL: Duration = Duration::from_secs(10);

    fn capacity(value: usize) -> NonZeroUsize {
        NonZeroUsize::new(value).unwrap_or(NonZeroUsize::MIN)
    }

    #[test]
    fn blocked_recv_wakes_promptly_on_external_close() {
        let hub = EventHub::default();
        let subscription = hub
            .subscribe(capacity(4))
            .unwrap_or_else(|_| unreachable!("subscribing to a fresh hub cannot be poisoned"));
        let close = subscription.close_handle();
        let (done, observed) = mpsc::channel();
        let (entering, entered) = mpsc::channel();
        let worker = thread::spawn(move || {
            entering.send(()).ok();
            let outcome = subscription.recv();
            done.send(outcome).ok();
        });
        entered
            .recv_timeout(HARNESS_WALL)
            .unwrap_or_else(|_| panic!("receiver thread never started"));
        // Settle so the worker is INSIDE the condvar wait before close fires;
        // without this the close-before-block path returns via the wait
        // predicate alone and the notify wake is never exercised.
        thread::sleep(Duration::from_millis(100));
        close.close();
        let outcome = observed
            .recv_timeout(HARNESS_WALL)
            .unwrap_or_else(|_| panic!("blocked recv() never woke on external close"));
        assert!(
            matches!(outcome, Err(EventReceiveError::Closed)),
            "external close must surface the typed Closed error, got: {outcome:?}"
        );
        worker
            .join()
            .unwrap_or_else(|_| panic!("receiver thread panicked"));
    }

    #[test]
    fn no_subscriber_churn_constructs_zero_events() {
        use std::sync::atomic::Ordering;
        let hub = EventHub::default();
        let id = ComponentId::derive("frame.test", "zero-cost");
        for _ in 0..64 {
            hub.publish_transition(id, None, LifecycleState::Registered)
                .unwrap_or_else(|_| unreachable!("publish with no subscribers cannot be poisoned"));
        }
        // The F-5a R6 zero-cost choke point: with no live subscriber,
        // publication returns before ANY event is constructed — the
        // deterministic counter at the construction site stays zero across
        // churn.
        assert_eq!(
            hub.constructed.load(Ordering::Acquire),
            0,
            "zero live subscribers must cost zero event constructions"
        );
        // Companion isolation: the counter genuinely counts — the same
        // publish with one live subscriber constructs exactly one event.
        let subscription = hub
            .subscribe(capacity(4))
            .unwrap_or_else(|_| unreachable!("subscribing to a fresh hub cannot be poisoned"));
        hub.publish_transition(id, None, LifecycleState::Registered)
            .unwrap_or_else(|_| unreachable!("publish to a live subscriber cannot be poisoned"));
        assert_eq!(hub.constructed.load(Ordering::Acquire), 1);
        drop(subscription);
    }

    #[test]
    fn recv_delivers_a_published_event_without_any_close() {
        let hub = EventHub::default();
        let subscription = hub
            .subscribe(capacity(4))
            .unwrap_or_else(|_| unreachable!("subscribing to a fresh hub cannot be poisoned"));
        let id = ComponentId::derive("frame.test", "event-close");
        hub.publish_transition(id, None, LifecycleState::Registered)
            .unwrap_or_else(|_| unreachable!("publish to a live subscriber cannot be poisoned"));
        let event = subscription
            .recv()
            .unwrap_or_else(|error| panic!("recv with a queued event must succeed: {error}"));
        assert_eq!(event.component_id, id);
    }

    #[test]
    fn close_is_idempotent_and_still_typed_after_drop() {
        let hub = EventHub::default();
        let subscription = hub
            .subscribe(capacity(4))
            .unwrap_or_else(|_| unreachable!("subscribing to a fresh hub cannot be poisoned"));
        let close = subscription.close_handle();
        close.close();
        close.close();
        assert!(matches!(
            subscription.recv(),
            Err(EventReceiveError::Closed)
        ));
        drop(subscription);
        close.close();
    }

    #[test]
    fn close_handle_does_not_keep_the_queue_alive() {
        let hub = EventHub::default();
        let subscription = hub
            .subscribe(capacity(1))
            .unwrap_or_else(|_| unreachable!("subscribing to a fresh hub cannot be poisoned"));
        let close = subscription.close_handle();
        drop(subscription);
        // Publication after the subscription dropped must construct no event
        // for it; the weak handle upgrading to nothing proves the queue died.
        let id = ComponentId::derive("frame.test", "event-close-weak");
        hub.publish_transition(id, None, LifecycleState::Registered)
            .unwrap_or_else(|_| unreachable!("publish with no live subscriber is a no-op"));
        assert!(close.queue.upgrade().is_none());
    }
}