frame_core/registry/fragments.rs
1//! Lifecycle-owned live fragment content and the authoritative fragment
2//! snapshot (F-5a R2/R3).
3//!
4//! The declared fragment ID SET rides the registration payload and is static
5//! per incarnation; CONTENT is live through the one lifecycle-owned update
6//! operation here. Both surfaces are fenced by the registry-minted
7//! incarnation: stale work from a departed incarnation receives the typed
8//! held-vs-current refusal and can never mutate its replacement.
9
10use std::sync::atomic::Ordering;
11
12use crate::error::RegistryError;
13use crate::event::LifecycleState;
14use crate::fragment::{ComponentIncarnation, FragmentKey, FragmentRow};
15
16use super::ComponentRegistry;
17use super::support::{ComponentRecord, state};
18
19impl ComponentRegistry {
20 /// Prepares one committed entry into Running, BEFORE the transition
21 /// publishes: resets live fragment content from the declared batch (the
22 /// ID set is static per incarnation and content restarts from the
23 /// declaration — F-5a R2's atomic visibility), then mints the entry's
24 /// incarnation ordinal, so a snapshot taken after the transition always
25 /// reads the batch and ordinal belonging to this entry (F-6a R1).
26 ///
27 /// The mint continues this IDENTITY's count (never the record's: a
28 /// remove-reset count would let a replacement registration re-mint an
29 /// ordinal a stale handle still holds — the committed red evidence).
30 /// The identity's first ever incarnation is 1 — the A3 generation
31 /// floor; each re-entry is strictly higher; and unrelated components'
32 /// start order can never perturb it (R6 determinism).
33 pub(super) fn prepare_running_incarnation(
34 &self,
35 id: crate::component::ComponentId,
36 record: &ComponentRecord,
37 ) -> Result<(), RegistryError> {
38 {
39 let mut content = record
40 .fragment_content
41 .lock()
42 .map_err(|_| RegistryError::SynchronizationPoisoned)?;
43 content.clear();
44 for declaration in &record.meta.fragments {
45 content.insert(declaration.id.clone(), declaration.content.clone());
46 }
47 }
48 let ordinal = {
49 let mut history = self
50 .incarnation_history
51 .lock()
52 .map_err(|_| RegistryError::SynchronizationPoisoned)?;
53 let entry = history.entry(id).or_insert(0);
54 *entry += 1;
55 *entry
56 };
57 record.incarnation.store(ordinal, Ordering::Release);
58 Ok(())
59 }
60 /// Applies one atomic whole-content swap to a declared fragment of a
61 /// Running incarnation, then publishes the fragment-affecting news item
62 /// (the stream carries the news; the snapshot stays the authority —
63 /// F-5a R3).
64 ///
65 /// Validation is identical to registration: the new content is checked
66 /// against the registry's caller-supplied `max_fragment_bytes`. The
67 /// mid-flight-mutation race class is handled, not avoided: the swap and
68 /// its Running/incarnation fence commit under the record's content lock,
69 /// so an update racing stop, crash, or remove either commits while the
70 /// incarnation is still live (and is invisible the instant the
71 /// incarnation leaves Running) or loses cleanly with the typed
72 /// held-vs-current refusal.
73 ///
74 /// # Errors
75 ///
76 /// Refuses an unregistered component, an unconfigured or exceeded byte
77 /// limit, a non-Running component or stale incarnation (both the typed
78 /// held-vs-current shape), an undeclared fragment id, and poisoned
79 /// synchronization.
80 pub fn update_fragment_content(
81 &self,
82 key: &FragmentKey,
83 held: ComponentIncarnation,
84 content: Vec<u8>,
85 ) -> Result<(), RegistryError> {
86 let record = self.record(key.component_id)?;
87 let fragment = key.fragment_id.as_str().to_owned();
88 let Some(limit) = self.config.max_fragment_bytes else {
89 return Err(RegistryError::FragmentLimitUnconfigured {
90 component: key.component_id,
91 });
92 };
93 if content.len() > limit.get() {
94 return Err(RegistryError::FragmentContentOversize {
95 component: key.component_id,
96 fragment,
97 size: content.len(),
98 limit: limit.get(),
99 });
100 }
101 {
102 let mut live = record
103 .fragment_content
104 .lock()
105 .map_err(|_| RegistryError::SynchronizationPoisoned)?;
106 // Fence under the content lock so the committed swap is atomic
107 // against every snapshot: state first (a non-Running component
108 // has NO live incarnation), then held-vs-current, then the id.
109 if state(&record)? != LifecycleState::Running {
110 return Err(RegistryError::FragmentIncarnationStale {
111 component: key.component_id,
112 fragment,
113 held,
114 current: None,
115 });
116 }
117 let current = ComponentIncarnation::new(record.incarnation.load(Ordering::Acquire));
118 if held != current {
119 return Err(RegistryError::FragmentIncarnationStale {
120 component: key.component_id,
121 fragment,
122 held,
123 current: Some(current),
124 });
125 }
126 let Some(slot) = live.get_mut(&key.fragment_id) else {
127 return Err(RegistryError::UnknownFragment {
128 component: key.component_id,
129 fragment,
130 });
131 };
132 *slot = content;
133 }
134 // News AFTER the committed truth, never before: a subscriber woken
135 // by this item always finds the swap in the authority.
136 self.events
137 .publish_fragment_update(key.clone())
138 .map_err(|_| RegistryError::SynchronizationPoisoned)
139 }
140
141 /// Returns the authoritative all-components fragment snapshot: one row
142 /// per declared fragment of every component currently in `Running`,
143 /// carrying current content bytes, sorted by [`FragmentKey`] canonical
144 /// byte order (F-5a R3's linearizable authority query; assembly ordering
145 /// — the R4 tuple — belongs to the assembly crate, not here).
146 ///
147 /// Per-component atomicity holds by construction: `Running`-ness is read
148 /// once per record, the incarnation ordinal is the one minted by that
149 /// entry into `Running`, and the record's content lock is held while its
150 /// rows are collected — a snapshot therefore contains ALL of a
151 /// component's fragments with one incarnation and no torn content, or
152 /// none of them.
153 ///
154 /// # Errors
155 ///
156 /// Returns a synchronization failure if registry state was poisoned, and
157 /// surfaces a declared id missing from the live store as the typed
158 /// unknown-fragment breach (impossible by construction, loud if broken).
159 pub fn fragment_snapshot(&self) -> Result<Vec<FragmentRow>, RegistryError> {
160 let records = self
161 .records
162 .lock()
163 .map_err(|_| RegistryError::SynchronizationPoisoned)?;
164 let mut rows = Vec::new();
165 for record in records.values() {
166 if state(record)? != LifecycleState::Running {
167 continue;
168 }
169 let incarnation = ComponentIncarnation::new(record.incarnation.load(Ordering::Acquire));
170 let content = record
171 .fragment_content
172 .lock()
173 .map_err(|_| RegistryError::SynchronizationPoisoned)?;
174 for declaration in &record.meta.fragments {
175 let Some(bytes) = content.get(&declaration.id) else {
176 return Err(RegistryError::UnknownFragment {
177 component: record.meta.id,
178 fragment: declaration.id.as_str().to_owned(),
179 });
180 };
181 rows.push(FragmentRow {
182 key: FragmentKey {
183 component_id: record.meta.id,
184 fragment_id: declaration.id.clone(),
185 },
186 kind: declaration.kind.clone(),
187 order: declaration.order,
188 content: bytes.clone(),
189 incarnation,
190 });
191 }
192 }
193 rows.sort_by(|left, right| left.key.cmp(&right.key));
194 Ok(rows)
195 }
196}