frame-core 0.3.0

//! Lifecycle-owned live fragment content and the authoritative fragment
//! snapshot (F-5a R2/R3).
//!
//! The declared fragment ID SET rides the registration payload and is static
//! per incarnation; CONTENT is live through the one lifecycle-owned update
//! operation here. Both surfaces are fenced by the registry-minted
//! incarnation: stale work from a departed incarnation receives the typed
//! held-vs-current refusal and can never mutate its replacement.

use std::sync::atomic::Ordering;

use crate::error::RegistryError;
use crate::event::LifecycleState;
use crate::fragment::{ComponentIncarnation, FragmentKey, FragmentRow};

use super::ComponentRegistry;
use super::support::{ComponentRecord, state};

impl ComponentRegistry {
    /// Prepares one committed entry into Running, BEFORE the transition
    /// publishes: resets live fragment content from the declared batch (the
    /// ID set is static per incarnation and content restarts from the
    /// declaration — F-5a R2's atomic visibility), then mints the entry's
    /// incarnation ordinal, so a snapshot taken after the transition always
    /// reads the batch and ordinal belonging to this entry (F-6a R1).
    ///
    /// The mint continues this IDENTITY's count (never the record's: a
    /// remove-reset count would let a replacement registration re-mint an
    /// ordinal a stale handle still holds — the committed red evidence).
    /// The identity's first ever incarnation is 1 — the A3 generation
    /// floor; each re-entry is strictly higher; and unrelated components'
    /// start order can never perturb it (R6 determinism).
    pub(super) fn prepare_running_incarnation(
        &self,
        id: crate::component::ComponentId,
        record: &ComponentRecord,
    ) -> Result<(), RegistryError> {
        {
            let mut content = record
                .fragment_content
                .lock()
                .map_err(|_| RegistryError::SynchronizationPoisoned)?;
            content.clear();
            for declaration in &record.meta.fragments {
                content.insert(declaration.id.clone(), declaration.content.clone());
            }
        }
        let ordinal = {
            let mut history = self
                .incarnation_history
                .lock()
                .map_err(|_| RegistryError::SynchronizationPoisoned)?;
            let entry = history.entry(id).or_insert(0);
            *entry += 1;
            *entry
        };
        record.incarnation.store(ordinal, Ordering::Release);
        Ok(())
    }
    /// Applies one atomic whole-content swap to a declared fragment of a
    /// Running incarnation, then publishes the fragment-affecting news item
    /// (the stream carries the news; the snapshot stays the authority —
    /// F-5a R3).
    ///
    /// Validation is identical to registration: the new content is checked
    /// against the registry's caller-supplied `max_fragment_bytes`. The
    /// mid-flight-mutation race class is handled, not avoided: the swap and
    /// its Running/incarnation fence commit under the record's content lock,
    /// so an update racing stop, crash, or remove either commits while the
    /// incarnation is still live (and is invisible the instant the
    /// incarnation leaves Running) or loses cleanly with the typed
    /// held-vs-current refusal.
    ///
    /// # Errors
    ///
    /// Refuses an unregistered component, an unconfigured or exceeded byte
    /// limit, a non-Running component or stale incarnation (both the typed
    /// held-vs-current shape), an undeclared fragment id, and poisoned
    /// synchronization.
    pub fn update_fragment_content(
        &self,
        key: &FragmentKey,
        held: ComponentIncarnation,
        content: Vec<u8>,
    ) -> Result<(), RegistryError> {
        let record = self.record(key.component_id)?;
        let fragment = key.fragment_id.as_str().to_owned();
        let Some(limit) = self.config.max_fragment_bytes else {
            return Err(RegistryError::FragmentLimitUnconfigured {
                component: key.component_id,
            });
        };
        if content.len() > limit.get() {
            return Err(RegistryError::FragmentContentOversize {
                component: key.component_id,
                fragment,
                size: content.len(),
                limit: limit.get(),
            });
        }
        {
            let mut live = record
                .fragment_content
                .lock()
                .map_err(|_| RegistryError::SynchronizationPoisoned)?;
            // Fence under the content lock so the committed swap is atomic
            // against every snapshot: state first (a non-Running component
            // has NO live incarnation), then held-vs-current, then the id.
            if state(&record)? != LifecycleState::Running {
                return Err(RegistryError::FragmentIncarnationStale {
                    component: key.component_id,
                    fragment,
                    held,
                    current: None,
                });
            }
            let current = ComponentIncarnation::new(record.incarnation.load(Ordering::Acquire));
            if held != current {
                return Err(RegistryError::FragmentIncarnationStale {
                    component: key.component_id,
                    fragment,
                    held,
                    current: Some(current),
                });
            }
            let Some(slot) = live.get_mut(&key.fragment_id) else {
                return Err(RegistryError::UnknownFragment {
                    component: key.component_id,
                    fragment,
                });
            };
            *slot = content;
        }
        // News AFTER the committed truth, never before: a subscriber woken
        // by this item always finds the swap in the authority.
        self.events
            .publish_fragment_update(key.clone())
            .map_err(|_| RegistryError::SynchronizationPoisoned)
    }

    /// Returns the authoritative all-components fragment snapshot: one row
    /// per declared fragment of every component currently in `Running`,
    /// carrying current content bytes, sorted by [`FragmentKey`] canonical
    /// byte order (F-5a R3's linearizable authority query; assembly ordering
    /// — the R4 tuple — belongs to the assembly crate, not here).
    ///
    /// Per-component atomicity holds by construction: `Running`-ness is read
    /// once per record, the incarnation ordinal is the one minted by that
    /// entry into `Running`, and the record's content lock is held while its
    /// rows are collected — a snapshot therefore contains ALL of a
    /// component's fragments with one incarnation and no torn content, or
    /// none of them.
    ///
    /// # Errors
    ///
    /// Returns a synchronization failure if registry state was poisoned, and
    /// surfaces a declared id missing from the live store as the typed
    /// unknown-fragment breach (impossible by construction, loud if broken).
    pub fn fragment_snapshot(&self) -> Result<Vec<FragmentRow>, RegistryError> {
        let records = self
            .records
            .lock()
            .map_err(|_| RegistryError::SynchronizationPoisoned)?;
        let mut rows = Vec::new();
        for record in records.values() {
            if state(record)? != LifecycleState::Running {
                continue;
            }
            let incarnation = ComponentIncarnation::new(record.incarnation.load(Ordering::Acquire));
            let content = record
                .fragment_content
                .lock()
                .map_err(|_| RegistryError::SynchronizationPoisoned)?;
            for declaration in &record.meta.fragments {
                let Some(bytes) = content.get(&declaration.id) else {
                    return Err(RegistryError::UnknownFragment {
                        component: record.meta.id,
                        fragment: declaration.id.as_str().to_owned(),
                    });
                };
                rows.push(FragmentRow {
                    key: FragmentKey {
                        component_id: record.meta.id,
                        fragment_id: declaration.id.clone(),
                    },
                    kind: declaration.kind.clone(),
                    order: declaration.order,
                    content: bytes.clone(),
                    incarnation,
                });
            }
        }
        rows.sort_by(|left, right| left.key.cmp(&right.key));
        Ok(rows)
    }
}