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 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();
465 self.observer.begin_frame();
466 let (root, commands, side_effects, requested_root_render, compact_applier) =
467 composer.install(|composer| {
468 let (_, outcome) = composer.try_with_slot_host_pass(
469 Rc::clone(host),
470 crate::slot::SlotPassMode::Recompose,
471 |composer| {
472 for scope in scopes {
473 if let Some(threshold_ms) = scope_telemetry_threshold_ms {
474 let start = Instant::now();
475 composer.recompose_group(scope);
476 let elapsed_ms = start.elapsed().as_secs_f64() * 1000.0;
477 if elapsed_ms >= threshold_ms {
478 eprintln!(
479 "[recompose-scope-telemetry] scope_id={} label={:?} elapsed_ms={elapsed_ms:.3} invalidation_sources={:?}",
480 scope.id(),
481 debug_scope_label(scope.id()),
482 debug_scope_invalidation_sources(scope.id())
483 );
484 }
485 } else {
486 composer.recompose_group(scope);
487 }
488 }
489 },
490 )?;
491 let root = composer.root();
492 let commands = composer.take_commands();
493 let side_effects = composer.take_side_effects();
494 let requested_root_render = composer.take_root_render_request();
495 Ok((
496 root,
497 commands,
498 side_effects,
499 requested_root_render,
500 outcome.compacted,
501 ))
502 })?;
503 self.record_pass_stats(&commands, &side_effects);
504 self.apply_commands_and_updates_for_host(host, &runtime_handle, commands)?;
505 if compact_applier {
506 self.applier.compact();
507 self.applier.borrow_dyn().clear_recycled_nodes();
508 }
509 if root.is_some() {
510 self.root = root;
511 }
512 if requested_root_render {
513 self.root_render_requested = true;
514 }
515 side_effects
516 };
517 runtime_handle.drain_ui();
518 for effect in side_effects {
519 effect();
520 }
521 runtime_handle.drain_ui();
522 self.maybe_dump_slot_table("recompose_pass");
523 if self.root_render_requested {
524 for (_, remaining_scopes) in scope_groups.iter().skip(host_group_index + 1) {
525 for scope in remaining_scopes {
526 runtime_handle.requeue_invalid_scope(scope.id(), scope.downgrade());
527 }
528 }
529 break;
530 }
531 host_group_index += 1;
532 }
533 if self.root_render_requested {
534 break;
535 }
536 }
537 self.finalize_runtime_state();
538 Ok(did_recompose)
539 }
540
541 pub fn process_invalid_scopes(&mut self) -> Result<bool, NodeError> {
542 self.process_invalid_scopes_until_root_request()
543 }
544
545 pub fn flush_pending_node_updates(&mut self) -> Result<(), NodeError> {
546 let updates = self.runtime_handle().take_updates();
547 let mut applier = self.applier.borrow_dyn();
548 for update in updates {
549 update.apply(&mut *applier)?;
550 }
551 Ok(())
552 }
553}
554
555impl<A: Applier + 'static> Drop for Composition<A> {
556 fn drop(&mut self) {
557 self.observer.stop();
558 }
559}