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
25pub 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 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 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 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 pub fn should_render(&self) -> bool {
316 self.root_render_requested || self.runtime.needs_frame() || self.runtime.has_updates()
317 }
318
319 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 self.dispose_forgotten_movables()?;
403 let Some(scopes) = live_invalidated_scopes(&runtime_handle) else {
404 break;
405 };
406 if scopes.is_empty() {
407 continue;
408 }
409 did_recompose = true;
410 let runtime_clone = runtime_handle.clone();
411 let root_host = self.slots_host();
412 let mut scope_groups: Vec<(Rc<SlotsHost>, Vec<RecomposeScope>)> = Vec::new();
413 let mut scope_group_index: HashMap<usize, usize> = HashMap::default();
414 for scope in scopes {
415 let host = scope
416 .slots_runtime_state()
417 .and_then(|state| {
418 scope
419 .slots_storage_key()
420 .and_then(|storage_key| state.host_for_storage_key(storage_key))
421 })
422 .or_else(|| {
423 scope.slots_storage_key().and_then(|storage_key| {
424 self.composer_state.host_for_storage_key(storage_key)
425 })
426 })
427 .unwrap_or_else(|| Rc::clone(&root_host));
428 let host_key = host.storage_key();
429 if let Some(index) = scope_group_index.get(&host_key).copied() {
430 scope_groups[index].1.push(scope);
431 } else {
432 scope_group_index.insert(host_key, scope_groups.len());
433 scope_groups.push((host, vec![scope]));
434 }
435 }
436 let mut host_group_index = 0usize;
437 while host_group_index < scope_groups.len() {
438 let (host, scopes) = &scope_groups[host_group_index];
439 let scope_telemetry_threshold_ms = recompose_scope_telemetry_threshold_ms();
440 let shared_state = host
441 .runtime_state()
442 .or_else(|| scopes.first().and_then(RecomposeScope::slots_runtime_state))
443 .unwrap_or_else(|| Rc::clone(&self.composer_state));
444 let side_effects = {
445 let _teardown = runtime::enter_state_teardown_scope();
446 let composer = Composer::new_with_shared_state(
447 shared_state,
448 Rc::clone(host),
449 self.applier_host(),
450 runtime_clone.clone(),
451 self.observer.clone(),
452 self.root,
453 );
454 composer.parent_stack().clear();
455 self.observer.begin_frame();
456 let (root, commands, side_effects, requested_root_render, compact_applier) =
457 composer.install(|composer| {
458 let (_, outcome) = composer.try_with_slot_host_pass(
459 Rc::clone(host),
460 crate::slot::SlotPassMode::Recompose,
461 |composer| {
462 for scope in scopes {
463 if let Some(threshold_ms) = scope_telemetry_threshold_ms {
464 let start = Instant::now();
465 composer.recompose_group(scope);
466 let elapsed_ms = start.elapsed().as_secs_f64() * 1000.0;
467 if elapsed_ms >= threshold_ms {
468 eprintln!(
469 "[recompose-scope-telemetry] scope_id={} label={:?} elapsed_ms={elapsed_ms:.3} invalidation_sources={:?}",
470 scope.id(),
471 debug_scope_label(scope.id()),
472 debug_scope_invalidation_sources(scope.id())
473 );
474 }
475 } else {
476 composer.recompose_group(scope);
477 }
478 }
479 },
480 )?;
481 let root = composer.root();
482 let commands = composer.take_commands();
483 let side_effects = composer.take_side_effects();
484 let requested_root_render = composer.take_root_render_request();
485 Ok((
486 root,
487 commands,
488 side_effects,
489 requested_root_render,
490 outcome.compacted,
491 ))
492 })?;
493 self.record_pass_stats(&commands, &side_effects);
494 self.apply_commands_and_updates_for_host(host, &runtime_handle, commands)?;
495 if compact_applier {
496 self.applier.compact();
497 self.applier.borrow_dyn().clear_recycled_nodes();
498 }
499 if root.is_some() {
500 self.root = root;
501 }
502 if requested_root_render {
503 self.root_render_requested = true;
504 }
505 side_effects
506 };
507 runtime_handle.drain_ui();
508 for effect in side_effects {
509 effect();
510 }
511 runtime_handle.drain_ui();
512 self.maybe_dump_slot_table("recompose_pass");
513 if self.root_render_requested {
514 for (_, remaining_scopes) in scope_groups.iter().skip(host_group_index + 1) {
515 for scope in remaining_scopes {
516 runtime_handle.requeue_invalid_scope(scope.id(), scope.downgrade());
517 }
518 }
519 break;
520 }
521 host_group_index += 1;
522 }
523 if self.root_render_requested {
524 break;
525 }
526 }
527 self.finalize_runtime_state();
528 Ok(did_recompose)
529 }
530
531 pub fn process_invalid_scopes(&mut self) -> Result<bool, NodeError> {
532 self.process_invalid_scopes_until_root_request()
533 }
534
535 fn dispose_forgotten_movables(&mut self) -> Result<(), NodeError> {
536 let runtime_handle = self.runtime_handle();
537 let ids = runtime_handle.take_forgotten_movables();
538 if ids.is_empty() {
539 return Ok(());
540 }
541 let host = self.slots_host();
542 let composer = Composer::new_with_shared_state(
543 Rc::clone(&self.composer_state),
544 Rc::clone(&host),
545 self.applier_host(),
546 runtime_handle.clone(),
547 self.observer.clone(),
548 self.root,
549 );
550 let commands = composer.install(|composer| {
551 composer.forget_movables(&ids)?;
552 Ok::<_, NodeError>(composer.take_commands())
553 })?;
554 self.apply_commands_and_updates_for_host(&host, &runtime_handle, commands)
555 }
556}
557
558fn live_invalidated_scopes(runtime_handle: &RuntimeHandle) -> Option<Vec<RecomposeScope>> {
559 let pending = runtime_handle.take_invalidated_scopes();
560 if pending.is_empty() {
561 return None;
562 }
563 let mut scopes = Vec::with_capacity(pending.len());
564 for (id, weak) in pending {
565 if let Some(inner) = weak.upgrade() {
566 scopes.push(RecomposeScope { inner });
567 } else {
568 runtime_handle.mark_scope_recomposed(id);
569 }
570 }
571 Some(scopes)
572}
573
574impl<A: Applier + 'static> Composition<A> {
575 pub fn flush_pending_node_updates(&mut self) -> Result<(), NodeError> {
576 let updates = self.runtime_handle().take_updates();
577 let mut applier = self.applier.borrow_dyn();
578 for update in updates {
579 update.apply(&mut *applier)?;
580 }
581 Ok(())
582 }
583}
584
585impl<A: Applier + 'static> Drop for Composition<A> {
586 fn drop(&mut self) {
587 self.observer.stop();
588 }
589}