cranpose_core/
composition.rs

1use std::rc::Rc;
2
3use web_time::Instant;
4
5use crate::{
6    Applier, ApplierGuard, ApplierHost, CommandQueue, Composer, CompositionPassDebugStats,
7    ConcreteApplierHost, DefaultScheduler, Key, NodeError, NodeId, RecomposeScope, RetentionPolicy,
8    Runtime, RuntimeHandle, ScopeId, SlotDebugSnapshot, SlotTable, SlotTableDebugStats, SlotsHost,
9    SnapshotStateObserver, collections::map::HashMap, debug_scope_invalidation_sources,
10    debug_scope_label, runtime, scheduler_ref, snapshot_state_observer,
11};
12
13pub struct Composition<A: Applier + 'static> {
14    pub(crate) composer_state: Rc<crate::composer::ComposerRuntimeState>,
15    pub(crate) slots: Rc<SlotsHost>,
16    pub(crate) applier: Rc<ConcreteApplierHost<A>>,
17    pub(crate) runtime: Runtime,
18    pub(crate) observer: SnapshotStateObserver,
19    pub(crate) root: Option<NodeId>,
20    pub(crate) root_key: Option<Key>,
21    pub(crate) root_render_requested: bool,
22    pub(crate) last_pass_stats: CompositionPassDebugStats,
23}
24
25/// Upper bound on chained root-render replays and scope-recomposition rounds.
26///
27/// Each root render clears `root_render_requested` but may re-raise it if a
28/// recompose pass inside `render()` promotes a scope callback to the root
29/// (see `Composer::recompose_group` in recompose.rs — callbacks that cannot
30/// run invalidate their `callback_promotion_target`, and if no ancestor can
31/// absorb the callback, `request_root_render()` is called). Each promotion
32/// walks up one parent scope, so natural convergence is bounded by the
33/// composition depth. The same invariant bounds `process_invalid_scopes`:
34/// recomposing a scope may invalidate others, but the chain must terminate.
35///
36/// This constant is a safety net for reentrant-render bugs. Exceeding it
37/// trips a `debug_assert!` in dev/test builds (loud failure so regressions
38/// are caught immediately) and falls back to a break + `log::error!` in
39/// release builds so end users do not see the UI thread panic.
40pub const ROOT_RENDER_REPLAY_LIMIT: usize = 100;
41
42fn recompose_scope_telemetry_threshold_ms() -> Option<f64> {
43    std::env::var("CRANPOSE_RECOMPOSE_SCOPE_TELEMETRY_MS")
44        .ok()
45        .and_then(|value| value.parse::<f64>().ok())
46        .filter(|value| value.is_finite() && *value >= 0.0)
47}
48
49impl<A: Applier + 'static> Composition<A> {
50    pub fn new(applier: A) -> Self {
51        Self::with_runtime(applier, Runtime::new(scheduler_ref(DefaultScheduler)))
52    }
53
54    pub fn with_runtime(applier: A, runtime: Runtime) -> Self {
55        let composer_state = Rc::new(crate::composer::ComposerRuntimeState::default());
56        let slots = Rc::new(SlotsHost::new(SlotTable::new()));
57        let applier = Rc::new(ConcreteApplierHost::new(applier));
58        let observer_handle = runtime.handle();
59        let observer = SnapshotStateObserver::new(move |callback| {
60            observer_handle.enqueue_ui_task(callback);
61        });
62        observer.start();
63        Self {
64            composer_state,
65            slots,
66            applier,
67            runtime,
68            observer,
69            root: None,
70            root_key: None,
71            root_render_requested: false,
72            last_pass_stats: CompositionPassDebugStats::default(),
73        }
74    }
75
76    /// Returns the root group key captured from the most recent `render()` call,
77    /// or `None` before the first render.
78    pub fn root_key(&self) -> Option<Key> {
79        self.root_key
80    }
81
82    pub fn set_retention_policy(&self, policy: RetentionPolicy) {
83        self.composer_state.set_retention_policy(policy);
84    }
85
86    fn slots_host(&self) -> Rc<SlotsHost> {
87        Rc::clone(&self.slots)
88    }
89
90    fn applier_host(&self) -> Rc<dyn ApplierHost> {
91        self.applier.clone()
92    }
93
94    fn reset_last_pass_stats(&mut self) {
95        self.last_pass_stats = CompositionPassDebugStats::default();
96    }
97
98    fn maybe_dump_slot_table(&self, label: &str) {
99        if !crate::env_flag!("COMPOSE_DEBUG_SLOT_TABLE") {
100            return;
101        }
102        eprintln!(
103            "[COMPOSE_DEBUG_SLOT_TABLE] {label}\n{:#?}",
104            self.debug_slot_snapshot()
105        );
106    }
107
108    pub fn take_root_render_request(&mut self) -> bool {
109        std::mem::take(&mut self.root_render_requested)
110    }
111
112    pub fn request_root_render(&mut self) {
113        self.root_render_requested = true;
114        self.runtime.handle().schedule();
115    }
116
117    fn record_pass_stats(
118        &mut self,
119        commands: &CommandQueue,
120        side_effects: &Vec<Box<dyn FnOnce()>>,
121    ) {
122        self.last_pass_stats.commands_len = self.last_pass_stats.commands_len.max(commands.len());
123        self.last_pass_stats.commands_cap =
124            self.last_pass_stats.commands_cap.max(commands.capacity());
125        self.last_pass_stats.command_payload_len_bytes = self
126            .last_pass_stats
127            .command_payload_len_bytes
128            .max(commands.payload_len_bytes());
129        self.last_pass_stats.command_payload_cap_bytes = self
130            .last_pass_stats
131            .command_payload_cap_bytes
132            .max(commands.payload_capacity_bytes());
133        self.last_pass_stats.sync_children_len = self
134            .last_pass_stats
135            .sync_children_len
136            .max(commands.sync_children.len());
137        self.last_pass_stats.sync_children_cap = self
138            .last_pass_stats
139            .sync_children_cap
140            .max(commands.sync_children.capacity());
141        self.last_pass_stats.sync_child_ids_len = self
142            .last_pass_stats
143            .sync_child_ids_len
144            .max(commands.sync_child_ids.len());
145        self.last_pass_stats.sync_child_ids_cap = self
146            .last_pass_stats
147            .sync_child_ids_cap
148            .max(commands.sync_child_ids.capacity());
149        self.last_pass_stats.side_effects_len = self
150            .last_pass_stats
151            .side_effects_len
152            .max(side_effects.len());
153        self.last_pass_stats.side_effects_cap = self
154            .last_pass_stats
155            .side_effects_cap
156            .max(side_effects.capacity());
157    }
158
159    fn finalize_runtime_state(&mut self) {
160        let runtime_handle = self.runtime_handle();
161        self.observer.prune_dead_scopes();
162        if !self.runtime.has_updates()
163            && !runtime_handle.has_invalid_scopes()
164            && !runtime_handle.has_frame_callbacks()
165            && !runtime_handle.has_pending_ui()
166        {
167            self.runtime.set_needs_frame(false);
168        }
169    }
170
171    fn abandon_host_after_apply_failure(&mut self, host: &Rc<SlotsHost>) {
172        host.abandon_after_apply_failure();
173        if Rc::ptr_eq(host, &self.slots) {
174            self.root = None;
175        }
176        self.root_render_requested = true;
177        self.finalize_runtime_state();
178    }
179
180    fn apply_commands_and_updates_for_host(
181        &mut self,
182        host: &Rc<SlotsHost>,
183        runtime_handle: &RuntimeHandle,
184        commands: CommandQueue,
185    ) -> Result<(), NodeError> {
186        let result = {
187            let mut applier = self.applier.borrow_dyn();
188            let mut result = commands.apply(&mut *applier);
189            if result.is_ok() {
190                for update in runtime_handle.take_updates() {
191                    if let Err(err) = update.apply(&mut *applier) {
192                        result = Err(err);
193                        break;
194                    }
195                }
196            }
197            result
198        };
199        if result.is_err() {
200            self.abandon_host_after_apply_failure(host);
201        }
202        result
203    }
204
205    fn render_root_pass(&mut self, key: Key, content: &mut dyn FnMut()) -> Result<(), NodeError> {
206        self.root_key = Some(key);
207        self.root_render_requested = false;
208        let runtime_handle = self.runtime_handle();
209        runtime_handle.drain_ui();
210        let side_effects = {
211            let _teardown = runtime::enter_state_teardown_scope();
212            let composer = Composer::new_with_shared_state(
213                Rc::clone(&self.composer_state),
214                Rc::clone(&self.slots),
215                self.applier.clone(),
216                runtime_handle.clone(),
217                self.observer.clone(),
218                self.root,
219            );
220            self.observer.begin_frame();
221            let (root, commands, side_effects, compact_applier) = composer.install(|composer| {
222                let (_, outcome) = composer.try_with_slot_host_pass(
223                    Rc::clone(&self.slots),
224                    crate::slot::SlotPassMode::Compose,
225                    |composer| composer.with_group(key, |_| content()),
226                )?;
227                let root = composer.root();
228                let commands = composer.take_commands();
229                let side_effects = composer.take_side_effects();
230                Ok((root, commands, side_effects, outcome.compacted))
231            })?;
232            self.record_pass_stats(&commands, &side_effects);
233            self.apply_commands_and_updates_for_host(
234                &Rc::clone(&self.slots),
235                &runtime_handle,
236                commands,
237            )?;
238            if compact_applier {
239                self.applier.compact();
240                self.applier.borrow_dyn().clear_recycled_nodes();
241            }
242
243            self.root = root;
244            side_effects
245        };
246        runtime_handle.drain_ui();
247        for effect in side_effects {
248            effect();
249        }
250        runtime_handle.drain_ui();
251        self.maybe_dump_slot_table("root_render_pass");
252        Ok(())
253    }
254
255    fn reconcile_with_content(
256        &mut self,
257        key: Key,
258        content: &mut dyn FnMut(),
259    ) -> Result<bool, NodeError> {
260        self.root_key = Some(key);
261        let mut did_work = false;
262        let mut root_render_replays = 0usize;
263        loop {
264            did_work |= self.process_invalid_scopes_until_root_request()?;
265            if !self.take_root_render_request() {
266                return Ok(did_work);
267            }
268
269            root_render_replays += 1;
270            if root_render_replays > ROOT_RENDER_REPLAY_LIMIT {
271                log::error!(
272                    "root render replay looped past {ROOT_RENDER_REPLAY_LIMIT} iterations; breaking to keep UI responsive"
273                );
274                return Err(NodeError::RecompositionLimitExceeded {
275                    operation: "root render replay",
276                    limit: ROOT_RENDER_REPLAY_LIMIT,
277                });
278            }
279
280            self.render_root_pass(key, content)?;
281            did_work = true;
282        }
283    }
284
285    pub fn render(&mut self, key: Key, mut content: impl FnMut()) -> Result<(), NodeError> {
286        self.reset_last_pass_stats();
287        self.render_root_pass(key, &mut content)?;
288        let _ = self.process_invalid_scopes()?;
289        Ok(())
290    }
291
292    /// Perform a root render and continue replaying any resulting root-render
293    /// requests until the composition reaches a stable fixpoint.
294    pub fn render_stable(&mut self, key: Key, mut content: impl FnMut()) -> Result<(), NodeError> {
295        self.reset_last_pass_stats();
296        self.render_root_pass(key, &mut content)?;
297        let _ = self.reconcile_with_content(key, &mut content)?;
298        Ok(())
299    }
300
301    /// Process invalid scopes and any resulting root-render requests until the
302    /// composition reaches a stable fixpoint for the supplied root content.
303    pub fn reconcile(&mut self, key: Key, mut content: impl FnMut()) -> Result<bool, NodeError> {
304        self.reconcile_with_content(key, &mut content)
305    }
306
307    /// Returns true if composition needs to process invalid scopes (recompose).
308    ///
309    /// This checks both:
310    /// - `has_updates()`: composition scopes that were invalidated by state changes
311    /// - `needs_frame()`: animation callbacks that may have pending work
312    ///
313    /// Note: For scroll performance, ensure scroll state changes use `Cell<T>` instead
314    /// of `MutableState<T>` to avoid triggering recomposition on every scroll frame.
315    pub fn should_render(&self) -> bool {
316        self.root_render_requested || self.runtime.needs_frame() || self.runtime.has_updates()
317    }
318
319    /// Whether any composition scope is actually invalid, i.e. whether running
320    /// the composable tree could produce a different result than last time.
321    ///
322    /// Deliberately excludes [`Runtime::needs_frame`], which [`Self::should_render`]
323    /// includes. An armed frame callback means the *runtime* owes someone a
324    /// tick - a future to resume, a dispatcher to drain - and every app with a
325    /// game loop or a polling effect has one armed at all times. Re-running the
326    /// composition for it recomposes a tree that nothing invalidated. Callers
327    /// deciding "should I tick" want [`Self::should_render`]; callers deciding
328    /// "should I recompose" want this.
329    ///
330    /// It is only meaningful *after* the frame callbacks have been drained: a
331    /// callback that writes state invalidates its readers as it runs, so the
332    /// answer is a question about work already discovered, not work still to
333    /// come.
334    pub fn should_recompose(&self) -> bool {
335        self.root_render_requested || self.runtime.has_updates()
336    }
337
338    pub fn runtime_handle(&self) -> RuntimeHandle {
339        self.runtime.handle()
340    }
341
342    pub fn applier_mut(&mut self) -> ApplierGuard<'_, A> {
343        ApplierGuard::new(self.applier.borrow_typed())
344    }
345
346    pub fn root(&self) -> Option<NodeId> {
347        self.root
348    }
349
350    pub fn debug_dump_slot_table_groups(&self) -> Vec<(usize, Key, Option<ScopeId>, usize)> {
351        self.slots.borrow().debug_dump_groups()
352    }
353
354    pub fn debug_dump_slot_entries(&self) -> Vec<crate::SlotDebugEntry> {
355        self.slots.borrow().debug_dump_slot_entries()
356    }
357
358    pub fn slot_table_heap_bytes(&self) -> usize {
359        self.slots.borrow().heap_bytes()
360    }
361
362    pub fn debug_slot_table_stats(&self) -> SlotTableDebugStats {
363        self.slots.debug_stats()
364    }
365
366    pub fn debug_slot_snapshot(&self) -> SlotDebugSnapshot {
367        self.slots.debug_snapshot()
368    }
369
370    pub fn debug_observer_stats(&self) -> snapshot_state_observer::SnapshotStateObserverDebugStats {
371        self.observer.debug_stats()
372    }
373
374    pub fn debug_last_pass_stats(&self) -> CompositionPassDebugStats {
375        self.last_pass_stats
376    }
377
378    #[cfg(test)]
379    pub(crate) fn debug_validate_slots(&self) -> Result<(), crate::slot::SlotInvariantError> {
380        let table = self.slots.borrow();
381        table.validate()?;
382        self.composer_state
383            .validate_host_retention(self.slots.as_ref(), &table)
384    }
385
386    fn process_invalid_scopes_until_root_request(&mut self) -> Result<bool, NodeError> {
387        let runtime_handle = self.runtime_handle();
388        let mut did_recompose = false;
389        let mut loop_count = 0;
390        loop {
391            loop_count += 1;
392            if loop_count > ROOT_RENDER_REPLAY_LIMIT {
393                log::error!(
394                    "process_invalid_scopes looped past {ROOT_RENDER_REPLAY_LIMIT} iterations; breaking to keep UI responsive"
395                );
396                return Err(NodeError::RecompositionLimitExceeded {
397                    operation: "process_invalid_scopes",
398                    limit: ROOT_RENDER_REPLAY_LIMIT,
399                });
400            }
401            runtime_handle.drain_ui();
402            let pending = runtime_handle.take_invalidated_scopes();
403            if pending.is_empty() {
404                break;
405            }
406            let mut scopes = Vec::new();
407            for (id, weak) in pending {
408                if let Some(inner) = weak.upgrade() {
409                    scopes.push(RecomposeScope { inner });
410                } else {
411                    runtime_handle.mark_scope_recomposed(id);
412                }
413            }
414            if scopes.is_empty() {
415                continue;
416            }
417            did_recompose = true;
418            let runtime_clone = runtime_handle.clone();
419            let root_host = self.slots_host();
420            let mut scope_groups: Vec<(Rc<SlotsHost>, Vec<RecomposeScope>)> = Vec::new();
421            let mut scope_group_index: HashMap<usize, usize> = HashMap::default();
422            for scope in scopes {
423                let host = scope
424                    .slots_runtime_state()
425                    .and_then(|state| {
426                        scope
427                            .slots_storage_key()
428                            .and_then(|storage_key| state.host_for_storage_key(storage_key))
429                    })
430                    .or_else(|| {
431                        scope.slots_storage_key().and_then(|storage_key| {
432                            self.composer_state.host_for_storage_key(storage_key)
433                        })
434                    })
435                    .unwrap_or_else(|| Rc::clone(&root_host));
436                let host_key = host.storage_key();
437                if let Some(index) = scope_group_index.get(&host_key).copied() {
438                    scope_groups[index].1.push(scope);
439                } else {
440                    scope_group_index.insert(host_key, scope_groups.len());
441                    scope_groups.push((host, vec![scope]));
442                }
443            }
444            let mut host_group_index = 0usize;
445            while host_group_index < scope_groups.len() {
446                let (host, scopes) = &scope_groups[host_group_index];
447                let scope_telemetry_threshold_ms = recompose_scope_telemetry_threshold_ms();
448                let shared_state = host
449                    .runtime_state()
450                    .or_else(|| scopes.first().and_then(RecomposeScope::slots_runtime_state))
451                    .unwrap_or_else(|| Rc::clone(&self.composer_state));
452                let side_effects = {
453                    let _teardown = runtime::enter_state_teardown_scope();
454                    let composer = Composer::new_with_shared_state(
455                        shared_state,
456                        Rc::clone(host),
457                        self.applier_host(),
458                        runtime_clone.clone(),
459                        self.observer.clone(),
460                        self.root,
461                    );
462                    composer.parent_stack().clear();
463                    self.observer.begin_frame();
464                    let (root, commands, side_effects, requested_root_render, compact_applier) =
465                        composer.install(|composer| {
466                            let (_, outcome) = composer.try_with_slot_host_pass(
467                                Rc::clone(host),
468                                crate::slot::SlotPassMode::Recompose,
469                                |composer| {
470                                    for scope in scopes {
471                                        if let Some(threshold_ms) = scope_telemetry_threshold_ms {
472                                            let start = Instant::now();
473                                            composer.recompose_group(scope);
474                                            let elapsed_ms = start.elapsed().as_secs_f64() * 1000.0;
475                                            if elapsed_ms >= threshold_ms {
476                                                eprintln!(
477                                                    "[recompose-scope-telemetry] scope_id={} label={:?} elapsed_ms={elapsed_ms:.3} invalidation_sources={:?}",
478                                                    scope.id(),
479                                                    debug_scope_label(scope.id()),
480                                                    debug_scope_invalidation_sources(scope.id())
481                                                );
482                                            }
483                                        } else {
484                                            composer.recompose_group(scope);
485                                        }
486                                    }
487                                },
488                            )?;
489                            let root = composer.root();
490                            let commands = composer.take_commands();
491                            let side_effects = composer.take_side_effects();
492                            let requested_root_render = composer.take_root_render_request();
493                            Ok((
494                                root,
495                                commands,
496                                side_effects,
497                                requested_root_render,
498                                outcome.compacted,
499                            ))
500                        })?;
501                    self.record_pass_stats(&commands, &side_effects);
502                    self.apply_commands_and_updates_for_host(host, &runtime_handle, commands)?;
503                    if compact_applier {
504                        self.applier.compact();
505                        self.applier.borrow_dyn().clear_recycled_nodes();
506                    }
507                    if root.is_some() {
508                        self.root = root;
509                    }
510                    if requested_root_render {
511                        self.root_render_requested = true;
512                    }
513                    side_effects
514                };
515                runtime_handle.drain_ui();
516                for effect in side_effects {
517                    effect();
518                }
519                runtime_handle.drain_ui();
520                self.maybe_dump_slot_table("recompose_pass");
521                if self.root_render_requested {
522                    for (_, remaining_scopes) in scope_groups.iter().skip(host_group_index + 1) {
523                        for scope in remaining_scopes {
524                            runtime_handle.requeue_invalid_scope(scope.id(), scope.downgrade());
525                        }
526                    }
527                    break;
528                }
529                host_group_index += 1;
530            }
531            if self.root_render_requested {
532                break;
533            }
534        }
535        self.finalize_runtime_state();
536        Ok(did_recompose)
537    }
538
539    pub fn process_invalid_scopes(&mut self) -> Result<bool, NodeError> {
540        self.process_invalid_scopes_until_root_request()
541    }
542
543    pub fn flush_pending_node_updates(&mut self) -> Result<(), NodeError> {
544        let updates = self.runtime_handle().take_updates();
545        let mut applier = self.applier.borrow_dyn();
546        for update in updates {
547            update.apply(&mut *applier)?;
548        }
549        Ok(())
550    }
551}
552
553impl<A: Applier + 'static> Drop for Composition<A> {
554    fn drop(&mut self) {
555        self.observer.stop();
556    }
557}
cranpose_core/composition.rs

cranpose_core/
composition.rs
